您可以使用SelectObject對目標文件執行SQL語句,返回執行結果。
無地域屬性的存儲空間不支持使用SelectObject。
背景信息
目前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兩種文件。
DOCUMENT是指整個文件是單一的JSON對象。
LINES表示整個文件由一行行的JSON對象組成,每一行是一個JSON對象(但整個文件本身并不是一個合法的JSON對象),行與行之間以換行分隔符隔開。OSS Select可以支持常見的\n,\r\n等分隔符,且無需用戶指定。
標準存儲類型和低頻訪問存儲類型的文件。歸檔存儲、冷歸檔存儲和深度冷歸檔存儲類型文件需要先執行解凍操作。
OSS完全托管加密、KMS托管主密鑰加密的文件。
支持的SQL語法
SQL語句: Select From Where
數據類型:string、int(64bit)、double(64bit), decimal(128bit) 、timestamp、bool
操作: 邏輯條件(AND,OR,NOT), 算術表達式(+-*/%), 比較操作(>,=, <, >=, <=, !=),String 操作 (LIKE, || )
重要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也提供了分片查詢的機制,包括以下兩種分片方式:
按行分片:常用的分片方式,然而對于稀疏數據來說,按行分片可能會導致分片時負載不均衡。
按Split分片:Split是OSS用于分片的一個概念,一個Split包含多行數據,每個Split的數據大小大致相等。
說明按Split分片比按行分片更加高效。
如果確定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掃描費用。預覽低頻訪問、歸檔存儲、冷歸檔存儲或者深度冷歸檔存儲類型文件時,會產生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查詢文件。
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);
}
}
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)
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。