背景資訊 目前Hadoop 3.0已經支援OSS在EMR上運行Spark、Hive、Presto等服務,同時阿里雲MaxCompute以及Data Lake Analytics均支援從OSS直接處理資料。
OSS提供的GetObject 介面決定了巨量資料平台只能把OSS資料全部下載到本地然後進行分析過濾,在很多查詢情境下浪費了大量頻寬和用戶端資源。
SelectObject 介面是對上述問題的解決方案。其核心思想是巨量資料平台將條件、Projection下推到OSS層,讓OSS做基本的過濾,從而只返回有用的資料。用戶端一方面可以減少網路頻寬,另一方面也減少了資料的處理量,從而節省了CPU和記憶體用來做其他更多的事情。這使得基於OSS的資料倉儲、資料分析成為一種更有吸引力的選擇。
費用說明 調用SelectObject 介面查詢資料時,按掃描的原檔案實際大小計費。更多資訊,請參見資料處理費用 。
支援的檔案類型 以下內容是對SelectObject 支援的檔案類型、支援的SQL文法等的詳細介紹。
RFC 4180標準的CSV(包括TSV等類CSV檔案,檔案的行資料行分隔符號以及Quote字元都可自訂)。
JSON檔案,且檔案編碼為UTF-8。JSON支援DOCUMENT和LINES兩種檔案。
標準儲存類型和低頻訪問儲存類型的檔案。Archive Storage、冷Archive Storage和深度冷Archive Storage類型檔案需要先執行解凍操作。
OSS完全託管加密、KMS託管主要金鑰加密的檔案。
支援的SQL文法 SQL語句: Select From Where
資料類型:string、int(64bit)、double(64bit), decimal(128bit) 、timestamp、bool
操作: 邏輯條件(AND,OR,NOT), 算術運算式(+-*/%), 比較操作(>,=, <, >=, <=, !=),String 操作 (LIKE, || )
支援的資料類型 OSS中的CSV資料預設都是String類型,您可以使用CAST函數實現資料轉換。
通過SQL查詢語句將_1和_2轉換為int的樣本:Select * from OSSOBject where cast (_1 as int) > cast(_2 as int)
同時,對於SelectObject 支援在Where條件中進行隱式轉換,例如下面語句中的第一列和第二列將被轉換成int:
Select _1 from ossobject where _1 + _2 > 100
對於JSON檔案,如果在SQL中未指定cast函數,則其類型根據JSON資料的實際類型而定,標準JSON內建的資料類型包括null、bool、int64、double、string等類型。
常見的SQL用例 常見的SQL用例包括CSV及JSON兩種。
CSV
應用情境
SQL語句
返回前10行資料
select * from ossobject limit 10
返回第1列和第3列的整數,並且第1列大於第3列
select _1, _3 from ossobject where cast(_1 as int) > cast(_3 as int)
返回第1列以'陳'開頭的記錄的個數(註:此處like後的中文需要用UTF-8編碼)
select count(*) from ossobject where _1 like '陳%'
返回所有第2列時間大於2018-08-09 11:30:25且第3列大於200的記錄
select * from ossobject where _2 > cast('2018-08-09 11:30:25' as timestamp) and _3 > 200
返回第2列浮點數的平均值,總和,最大值,最小值
select AVG(cast(_6 as double)), SUM(cast(_6 as double)), MAX(cast(_6 as double)), MIN(cast(_6 as double)) from ossobject
返回第1列和第3列串連的字串中以'Tom'為開頭以’Anderson‘結尾的所有記錄
select * from ossobject where (_1 || _3) like 'Tom%Anderson'
返回第1列能被3整除的所有記錄
select * from ossobject where (_1 % 3) = 0
返回第1列大小在1995到2012之間的所有記錄
select * from ossobject where _1 between 1995 and 2012
返回第5列值為N,M,G,L的所有記錄
select * from ossobject where _5 in ('N', 'M', 'G', 'L')
返回第2列乘以第3列比第5列大100以上的所有記錄
select * from ossobject where _2 * _3 > _5 + 100
JSON
假設JSON檔案如下:
{
"contacts":[
{
"firstName": "John",
"lastName": "Smith",
"isAlive": true,
"age": 27,
"address": {
"streetAddress": "21 2nd Street",
"city": "New York",
"state": "NY",
"postalCode": "10021-3100"
},
"phoneNumbers": [
{
"type": "home",
"number": "212 555-1234"
},
{
"type": "office",
"number": "646 555-4567"
},
{
"type": "mobile",
"number": "123 456-7890"
}
],
"children": [],
"spouse": null
},…… #此處省略其他類似的節點
]}
SQL用例如下:
應用情境
SQL語句
返回所有age是27的記錄
select * from ossobject.contacts[*] s where s.age = 27
返回所有的家庭電話
select s.number from ossobject.contacts[*].phoneNumbers[*] s where s.type = “home”
返回所有單身的記錄
select * from ossobject s where s.spouse is null
返回所有沒有孩子的記錄
select * from ossobject s where s.children[0] is null
說明 目前沒有專用的空數組的表示方法,用以上語句代替。
使用情境 SelectObject通常用於大檔案分區查詢、JSON檔案查詢、記錄檔分析等情境。
大檔案分區查詢
和GetObject 提供的基於Byte的分區下載類似,SelectObject 也提供了分區查詢的機制,包括以下兩種分區方式:
如果確定CSV檔案列中不包含分行符號,則基於Bytes的分區由於不需要建立Meta,其使用更為簡便。如果列中包含分行符號或者是JSON檔案時,則使用以下步驟:
調用CreateSelectObjectMeta API獲得該檔案的總的Split數。如果該檔案需要用SelectObject,則建議在查詢前非同步呼叫該介面,以節省掃描時間。
根據用戶端資源情況選擇合適的並發度n,用總的Split數除以並發度n得到每個分區查詢應該包含的Split個數。
在請求Body中用諸如split-range=1-20的形式進行分區查詢。
合并結果。
JSON檔案查詢
查詢JSON檔案時,在SQL的From語句中儘可能縮小From後的JSON Path範圍。
如下是JSON檔案樣本:
{
"contacts":[
{
"firstName": "John",
"lastName": "Smith",
"address": {
"streetAddress": "21 2nd Street",
"city": "New York",
"state": "NY",
"postalCode": "10021-3100"
},
"phoneNumbers": [
{
"type": "home",
"number": "212 555-1234"
},
{
"type": "office",
"number": "646 555-4567"
},
{
"type": "mobile",
"number": "123 456-7890"
}
]
}
]}
如果要尋找所有postalCode為10021開頭的streetAddress,SQL可以寫為 select s.address.streetAddress from ossobject.contacts[*] s where s.address.postalCode like '10021%'
或者select s.streetAddress from ossobject.contacts[*].address s where s.postalCode like '10021%'
由於select s.streetAddress from ossobject.contacts[*].address s where s.postalCode like '10021%'
的JSON Path更加精確,因此效能更優。
在JSON檔案中處理高精度浮點數
在JSON檔案中需要進行高精度浮點數的數值計算時,建議設定ParseJsonNumberAsString選項為true, 同時將該值cast成Decimal。比如一個屬性a值為123456789.123456789,用select s.a from ossobject s where cast(s.a as decimal) > 123456789.12345
就可以保持未經處理資料的精度不丟失。
操作步驟 使用OSS控制台
重要 通過控制台僅支援從128 MB以下的檔案中提取40 MB以下的資料記錄。
登入OSS管理主控台 。
單擊Bucket 列表 ,然後單擊目標Bucket名稱。
在左側導覽列,選擇文件管理 >檔案清單 。
在目標檔案右側的操作欄下,選擇。
在選取內容 面板,按以下說明設定各項參數。
參數
說明
檔案類型
僅支援CSV和JSON兩種檔案類型。
分隔字元
僅適用於CSV檔案。請選擇半形逗號(,)或自訂分隔字元。
標題列
僅適用於CSV檔案。請選擇檔案第一行是否包含欄位標題。
JSON格式符
僅適用於JSON檔案。請選擇您的JSON檔案對應的格式。
壓縮格式
選擇您當前的檔案是否為壓縮檔。目前壓縮檔僅支援GZIP檔案。
單擊顯示檔案預覽 。
重要 預覽標準儲存類型檔案時,會產生Select掃描費用。預覽低頻訪問、Archive Storage、冷Archive Storage或者深度冷Archive Storage類型檔案時,會產生Select掃描費用和資料取回費用。更多資訊,請參見資料處理費用 。
單擊下一步 ,輸入SQL語句並執行。
假設名為People 的CSV檔案有3列資料,分別是姓名 、公司 和年齡 。
如果想尋找年齡大於50歲,並且名字以Lora開頭的人(其中_1,_2,_3是列索引,代表第一列、第二列、第三列),可以執行以下SQL語句:
select * from ossobject where _1 like 'Lora*' and _3 > 50
如果想統計這個檔案有多少行,最大年齡與最小年齡是多少,可以執行以下SQL語句:
select count(*), max(cast(_3 as int)), min(cast(_3 as int)) from oss_object
查看執行結果。
使用阿里雲SDK 當前僅支援通過Java SDK和Python SDK查詢檔案。
Java import com.aliyun.oss.model.*;
import com.aliyun.oss.OSS;
import com.aliyun.oss.common.auth.*;
import com.aliyun.oss.OSSClientBuilder;
import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.InputStreamReader;
/**
* Examples of create select object metadata and select object.
*
*/
public class SelectObjectSample {
// yourEndpoint填寫Bucket所在地區對應的Endpoint。以華東1(杭州)為例,Endpoint填寫為https://oss-cn-hangzhou.aliyuncs.com。
private static String endpoint = "https://oss-cn-hangzhou.aliyuncs.com";
// 填寫Bucket名稱,例如examplebucket。
private static String bucketName = "examplebucket";
public static void main(String[] args) throws Exception {
// 從環境變數中擷取訪問憑證。運行本程式碼範例之前,請確保已設定環境變數OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
EnvironmentVariableCredentialsProvider credentialsProvider = CredentialsProviderFactory.newEnvironmentVariableCredentialsProvider();
// 建立OSSClient執行個體。
OSS ossClient = new OSSClientBuilder().build(endpoint, credentialsProvider);
// 填寫Object完整路徑後,根據SELECT語句查詢檔案中的資料。Object完整路徑中不能包含Bucket名稱。
// 填寫CSV格式的Object完整路徑。
selectCsvSample("test.csv", ossClient);
// 填寫JSON格式的Object完整路徑。
selectJsonSample("test.json", ossClient);
ossClient.shutdown();
}
private static void selectCsvSample(String key, OSS ossClient) throws Exception {
// 填寫上傳的內容。
String content = "name,school,company,age\r\n" +
"Lora Francis,School A,Staples Inc,27\r\n" +
"Eleanor Little,School B,\"Conectiv, Inc\",43\r\n" +
"Rosie Hughes,School C,Western Gas Resources Inc,44\r\n" +
"Lawrence Ross,School D,MetLife Inc.,24";
ossClient.putObject(bucketName, key, new ByteArrayInputStream(content.getBytes()));
SelectObjectMetadata selectObjectMetadata = ossClient.createSelectObjectMetadata(
new CreateSelectObjectMetadataRequest(bucketName, key)
.withInputSerialization(
new InputSerialization().withCsvInputFormat(
// 填寫內容中不同記錄之間的分隔字元,例如\r\n。
new CSVFormat().withHeaderInfo(CSVFormat.Header.Use).withRecordDelimiter("\r\n"))));
System.out.println(selectObjectMetadata.getCsvObjectMetadata().getTotalLines());
System.out.println(selectObjectMetadata.getCsvObjectMetadata().getSplits());
SelectObjectRequest selectObjectRequest =
new SelectObjectRequest(bucketName, key)
.withInputSerialization(
new InputSerialization().withCsvInputFormat(
new CSVFormat().withHeaderInfo(CSVFormat.Header.Use).withRecordDelimiter("\r\n")))
.withOutputSerialization(new OutputSerialization().withCsvOutputFormat(new CSVFormat()));
// 使用SELECT語句查詢第4列,值大於40的所有記錄。
selectObjectRequest.setExpression("select * from ossobject where _4 > 40");
OSSObject ossObject = ossClient.selectObject(selectObjectRequest);
// 讀取內容。
BufferedReader reader = new BufferedReader(new InputStreamReader(ossObject.getObjectContent()));
while (true) {
String line = reader.readLine();
if (line == null) {
break;
}
System.out.println(line);
}
reader.close();
ossClient.deleteObject(bucketName, key);
}
private static void selectJsonSample(String key, OSS ossClient) throws Exception {
// 填寫上傳的內容。
final String content = "{\n" +
"\t\"name\": \"Lora Francis\",\n" +
"\t\"age\": 27,\n" +
"\t\"company\": \"Staples Inc\"\n" +
"}\n" +
"{\n" +
"\t\"name\": \"Eleanor Little\",\n" +
"\t\"age\": 43,\n" +
"\t\"company\": \"Conectiv, Inc\"\n" +
"}\n" +
"{\n" +
"\t\"name\": \"Rosie Hughes\",\n" +
"\t\"age\": 44,\n" +
"\t\"company\": \"Western Gas Resources Inc\"\n" +
"}\n" +
"{\n" +
"\t\"name\": \"Lawrence Ross\",\n" +
"\t\"age\": 24,\n" +
"\t\"company\": \"MetLife Inc.\"\n" +
"}";
ossClient.putObject(bucketName, key, new ByteArrayInputStream(content.getBytes()));
SelectObjectRequest selectObjectRequest =
new SelectObjectRequest(bucketName, key)
.withInputSerialization(new InputSerialization()
.withCompressionType(CompressionType.NONE)
.withJsonInputFormat(new JsonFormat().withJsonType(JsonType.LINES)))
.withOutputSerialization(new OutputSerialization()
.withCrcEnabled(true)
.withJsonOutputFormat(new JsonFormat()))
.withExpression("select * from ossobject as s where s.age > 40"); // 使用SELECT語句查詢檔案中的資料。
OSSObject ossObject = ossClient.selectObject(selectObjectRequest);
// 讀取內容。
BufferedReader reader = new BufferedReader(new InputStreamReader(ossObject.getObjectContent()));
while (true) {
String line = reader.readLine();
if (line == null) {
break;
}
System.out.println(line);
}
reader.close();
ossClient.deleteObject(bucketName, key);
}
}
Python import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
def select_call_back(consumed_bytes, total_bytes = None):
print('Consumed Bytes:' + str(consumed_bytes) + '\n')
# 從環境變數中擷取訪問憑證。運行本程式碼範例之前,請確保已設定環境變數OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
# yourEndpoint填寫Bucket所在地區對應的Endpoint。以華東1(杭州)為例,Endpoint填寫為https://oss-cn-hangzhou.aliyuncs.com。
# 填寫Bucket名稱。
bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', 'examplebucket')
key = 'python_select.csv'
content = 'Tom Hanks,USA,45\r\n'*1024
filename = 'python_select.csv'
# 上傳CSV檔案。
bucket.put_object(key, content)
# Select API的參數。
csv_meta_params = {'RecordDelimiter': '\r\n'}
select_csv_params = {'CsvHeaderInfo': 'None',
'RecordDelimiter': '\r\n',
'LineRange': (500, 1000)}
csv_header = bucket.create_select_object_meta(key, csv_meta_params)
print(csv_header.rows)
print(csv_header.splits)
result = bucket.select_object(key, "select * from ossobject where _3 > 44", select_call_back, select_csv_params)
select_content = result.read()
print(select_content)
result = bucket.select_object_to_file(key, filename,
"select * from ossobject where _3 > 44", select_call_back, select_csv_params)
bucket.delete_object(key)
###JSON DOCUMENT
key = 'python_select.json'
content = "{\"contacts\":[{\"key1\":1,\"key2\":\"hello world1\"},{\"key1\":2,\"key2\":\"hello world2\"}]}"
filename = 'python_select.json'
# 上傳JSON DOCUMENT。
bucket.put_object(key, content)
select_json_params = {'Json_Type': 'DOCUMENT'}
result = bucket.select_object(key, "select s.key2 from ossobject.contacts[*] s where s.key1 = 1", None, select_json_params)
select_content = result.read()
print(select_content)
result = bucket.select_object_to_file(key, filename,
"select s.key2 from ossobject.contacts[*] s where s.key1 = 1", None, select_json_params)
bucket.delete_object(key)
###JSON LINES
key = 'python_select_lines.json'
content = "{\"key1\":1,\"key2\":\"hello world1\"}\n{\"key1\":2,\"key2\":\"hello world2\"}"
filename = 'python_select.json'
# 上傳JSON LINE。
bucket.put_object(key, content)
select_json_params = {'Json_Type': 'LINES'}
json_header = bucket.create_select_object_meta(key,select_json_params)
print(json_header.rows)
print(json_header.splits)
result = bucket.select_object(key, "select s.key2 from ossobject s where s.key1 = 1", None, select_json_params)
select_content = result.read()
print(select_content)
result = bucket.select_object_to_file(key, filename,
"select s.key2 from ossobject s where s.key1 = 1", None, select_json_params)
bucket.delete_object(key)
Go package main
import (
"fmt"
"github.com/aliyun/aliyun-oss-go-sdk/oss"
"io/ioutil"
"os"
)
func main() {
// 從環境變數中擷取訪問憑證。運行本程式碼範例之前,請確保已設定環境變數OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
provider, err := oss.NewEnvironmentVariableCredentialsProvider()
if err != nil {
fmt.Println("Error:", err)
os.Exit(-1)
}
// 建立OSSClient執行個體。
// yourEndpoint填寫Bucket對應的Endpoint,以華東1(杭州)為例,填寫為https://oss-cn-hangzhou.aliyuncs.com。其它Region請按實際情況填寫。
client, err := oss.New("yourEndpoint", "", "", oss.SetCredentialsProvider(&provider))
if err != nil {
fmt.Println("Error:", err)
os.Exit(-1)
}
// 填寫Bucket名稱,例如examplebucket。
bucket, err := client.Bucket("examplebucket")
if err != nil {
fmt.Println("Error:", err)
os.Exit(-1)
}
// 填寫Object完整路徑,完整路徑中不能包含Bucket名稱,例如exampledir/exampledata.csv。
key := "exampledir/exampledata.csv"
// 填寫本地CSV檔案的完整路徑,例如D:\\localpath\\exampledata.csv。
localCsvFile := "D:\\localpath\\exampledata.csv"
err = bucket.PutObjectFromFile(key, localCsvFile)
if err != nil {
fmt.Println("Error:", err)
os.Exit(-1)
}
selReq := oss.SelectRequest{}
// 使用SELECT語句查詢檔案中的資料。
selReq.Expression = `select * from ossobject`
body, err := bucket.SelectObject(key, selReq)
if err != nil {
fmt.Println("Error:", err)
os.Exit(-1)
}
// 讀取內容。
fc, err := ioutil.ReadAll(body)
if err != nil {
fmt.Println("Error:", err)
os.Exit(-1)
}
defer body.Close()
fmt.Println(string(fc))
}
使用REST API 如果您的程式自訂要求較高,您可以直接發起REST API請求。直接發起REST API請求需要手動編寫代碼計算簽名。更多資訊,請參見SelectObject 。