3.5. S3 选择操作
作为开发者,您可以运行 S3 选择以加快吞吐量。用户可以在没有介质器的情况下直接运行 S3 选择查询。
有三个 S3 选择工作流 - CSV、Apache Parquet (Parquet)和 JSON,它们为 CSV、Parquet 和 JSON 对象提供 S3 选择操作:
- CSV 文件以纯文本格式存储表格数据。文件的每一行都是数据记录。
- Parquet 是一个开源、面向列的数据文件格式,旨在有效数据存储和检索。它提供高效的数据压缩和编码方案,提高性能以批量处理复杂数据。Parquet 允许 S3 select-engine 跳过列和块,从而显著减少 IOPS (与 CSV 和 JSON 格式保持一致)。
- JSON 是一种格式结构。S3 选择引擎利用 JSON 读取器在 JSON 格式输入数据上使用 SQL 语句,从而扫描高度嵌套和复杂的 JSON 格式的数据。
例如,带有几GB 数据的 CSV、Parquet 或 JSON S3 对象,用户可以使用以下查询提取单个列,该列由另一列过滤:
Example
select customerid from s3Object where age>30 and age<65;
目前,S3 对象必须通过 Ceph 对象网关从 Ceph OSD 检索数据,然后才能过滤和提取数据。当对象较大且查询更为具体时,会提高性能。与 CSV 相比,Parquet 格式可以更有效地处理。
先决条件
- 一个正在运行的 Red Hat Ceph Storage 集群。
- RESTful 客户端。
- 创建的用户具有访问权限的 S3 用户。
3.5.1. S3 从对象中选择内容 复制链接链接已复制到粘贴板!
select 对象内容 API 通过结构化查询语言(SQL)过滤对象的内容。如需清单对象中应驻留的内容的说明,请参阅 AWS 系统管理器用户指南中的清单收集的元数据部分。清单内容会影响应针对该清单运行的查询类型。可能提供必要信息的 SQL 语句数量较大,但 S3 选择是一个类似于 SQL 的实用程序,因此不支持一些运算符,如 group-by 和 join。
对于 CSV,您必须将数据序列化格式指定为以逗号分隔的对象值,才能检索指定的内容。Parquet 没有分隔符,因为它采用二进制格式。Amazon Web Services (AWS)命令行界面(CLI)选择对象内容使用 CSV 或 Parquet 格式将对象数据解析到记录中,仅返回查询中指定的记录。
您必须为响应指定数据序列化格式。这个操作必须具有 s3:GetObject 权限。
-
InputSerialization元素描述正在查询的对象中数据的格式。对象可以是 CSV 或 Parquet 格式。 -
OutputSerialization元素是 AWS-CLI 用户客户端的一部分,描述了如何格式化输出数据。Ceph 为 AWS-CLI 实施服务器客户端,因此根据OutputSerialization(当前仅为 CSV)提供相同的输出。 -
InputSerialization的格式不需要与OutputSerialization的格式匹配。例如,您可以在InputSerialization中指定 Parquet,在OutputSerialization中指定 CSV。
语法
POST /BUCKET/KEY?select&select-type=2 HTTP/1.1\r\n
示例
POST /testbucket/sample1csv?select&select-type=2 HTTP/1.1\r\n
POST /testbucket/sample1parquet?select&select-type=2 HTTP/1.1\r\n
请求实体
Bucket- 描述
- 要从中选择对象内容的存储桶。
- Type
- 字符串
- 必需
- 是
键- 描述
- 对象密钥。
- 长度限制
- 最小长度为 1.
- Type
- 字符串
- 必需
- 是
SelectObjectContentRequest- 描述
- 选择对象内容请求参数的根级别标签。
- Type
- 字符串
- 必需
- 是
表达式- 描述
- 用于查询对象的表达式。
- Type
- 字符串
- 必需
- 是
ExpressionType- 描述
- 示例 SQL 提供的表达式的类型。
- Type
- 字符串
- 有效值
- SQL
- 必需
- 是
InputSerialization- 描述
- 描述正在查询的对象中数据的格式。
- Type
- 字符串
- 必需
- 是
OutputSerialization- 描述
- 以逗号分隔符和换行符返回的数据格式。
- Type
- 字符串
- 必需
- 是
响应实体
如果操作成功,服务会返回 HTTP 200 响应。服务以 XML 格式返回数据:
payload- 描述
- 有效负载参数的根级别标签。
- Type
- 字符串
- 必需
- 是
Records- 描述
- 记录事件。
- Type
- base64 编码的二进制数据对象
- 必需
- 否
Stats- 描述
- stats 事件。
- Type
- Long
- 必需
- 否
Ceph 对象网关支持以下响应:
示例
{:event-type,records} {:content-type,application/octet-stream} {:message-type,event}
语法(用于 CSV)
aws --endpoint-URL http://localhost:80 s3api select-object-content
--bucket BUCKET_NAME
--expression-type 'SQL'
--input-serialization
'{"CSV": {"FieldDelimiter": "," , "QuoteCharacter": "\"" , "RecordDelimiter" : "\n" , "QuoteEscapeCharacter" : "\\" , "FileHeaderInfo": "USE" }, "CompressionType": "NONE"}'
--output-serialization '{"CSV": {}}'
--key OBJECT_NAME.csv
--expression "select count(0) from s3object where int(_1)<10;" output.csv
示例(用于 CSV)
aws --endpoint-url http://localhost:80 s3api select-object-content
--bucket testbucket
--expression-type 'SQL'
--input-serialization
'{"CSV": {"FieldDelimiter": "," , "QuoteCharacter": "\"" , "RecordDelimiter" : "\n" , "QuoteEscapeCharacter" : "\\" , "FileHeaderInfo": "USE" }, "CompressionType": "NONE"}'
--output-serialization '{"CSV": {}}'
--key testobject.csv
--expression "select count(0) from s3object where int(_1)<10;" output.csv
语法(用于 Parquet)
aws --endpoint-url http://localhost:80 s3api select-object-content
--bucket BUCKET_NAME
--expression-type 'SQL'
--input-serialization
'{"Parquet": {}, {"CompressionType": "NONE"}'
--output-serialization '{"CSV": {}}'
--key OBJECT_NAME.parquet
--expression "select count(0) from s3object where int(_1)<10;" output.csv
示例(用于 Parquet)
aws --endpoint-url http://localhost:80 s3api select-object-content
--bucket testbucket
--expression-type 'SQL'
--input-serialization
'{"Parquet": {}, {"CompressionType": "NONE"}'
--output-serialization '{"CSV": {}}'
--key testobject.parquet
--expression "select count(0) from s3object where int(_1)<10;" output.csv
语法(用于 JSON)
aws --endpoint-URL http://localhost:80 s3api select-object-content
--bucket BUCKET_NAME
--expression-type 'SQL'
--input-serialization
'{"JSON": {"CompressionType": "NONE"}'
--output-serialization '{"CSV": {}}}'
--key OBJECT_NAME.json
--expression "select count(0) from s3object where int(_1)<10;" output.csv
示例(用于 JSON)
aws --endpoint-url http://localhost:80 s3api select-object-content
--bucket testbucket
--expression-type 'SQL'
--input-serialization
'{"JSON": {"CompressionType": "NONE"}'
--output-serialization '{"CSV": {}}}'
--key testobject.json
--expression "select count(0) from s3object where int(_1)<10;" output.csv
示例(如 BOTO3)
import pprint
import boto3
from botocore.exceptions import ClientError
def run_s3select(bucket,key,query,column_delim=",",row_delim="\n",quot_char='"',esc_char='\\',csv_header_info="NONE"):
s3 = boto3.client('s3',
endpoint_url=endpoint,
aws_access_key_id=access_key,
region_name=region_name,
aws_secret_access_key=secret_key)
result = ""
try:
r = s3.select_object_content(
Bucket=bucket,
Key=key,
ExpressionType='SQL',
InputSerialization = {"CSV": {"RecordDelimiter" : row_delim, "FieldDelimiter" : column_delim,"QuoteEscapeCharacter": esc_char, "QuoteCharacter": quot_char, "FileHeaderInfo": csv_header_info}, "CompressionType": "NONE"},
OutputSerialization = {"CSV": {}},
Expression=query,
RequestProgress = {"Enabled": progress})
except ClientError as c:
result += str(c)
return result
for event in r['Payload']:
if 'Records' in event:
result = ""
records = event['Records']['Payload'].decode('utf-8')
result += records
if 'Progress' in event:
print("progress")
pprint.pprint(event['Progress'],width=1)
if 'Stats' in event:
print("Stats")
pprint.pprint(event['Stats'],width=1)
if 'End' in event:
print("End")
pprint.pprint(event['End'],width=1)
return result
run_s3select(
"my_bucket",
"my_csv_object",
"select int(_1) as a1, int(_2) as a2 , (a1+a2) as a3 from s3object where a3>100 and a3<300;")
支持的功能
目前只支持 AWS s3 选择命令的一部分: