使用AWS S3 Select + boto3时遍历EventStream延迟过高如何优化？

我使用AWS S3 Select结合boto3从S3存储桶获取CSV数据块，查询返回的resp['Payload']是botocore.eventstream.EventStream对象。遍历该EventStream解码JSON响应并加载到Python列表时，耗时约4-5分钟，希望找到提速的方法。

原实现代码：

query="""SELECT * FROM s3object s WHERE "s"."xyz"='"""+num+"""' AND "s"."datestr">='"""+date_raw+"""' AND "s"."timestr">='"""+time_raw+"""' AND "s"."datestr"<='""" +date_raw_to+"""' AND "s"."timestr"<='"""+time_raw_to+"""'"""
        
print("query: ", query)
key="base/"+filepath

t1=time.time()
resp = s3_client.select_object_content(
    Bucket=bucket_name,
    Key=key, # file name or complete path with folder
    Expression=query, # query to execute on the CSV
    ExpressionType='SQL',
    InputSerialization={
        "CSV": {
            'FileHeaderInfo': 'USE',
        },
    },
    OutputSerialization={
        'JSON': {}
    },
)
print("resp: ", resp)
print("resp: ", type(resp)) # prints <class 'dict'>

print(type(resp['Payload'])) # prints <class 'botocore.eventstream.EventStream'>

t2=time.time()
print("time taken to get select response: ", t2-t1) # 2.6 seconds
all_records=[]
t1=time.time()
for event in resp['Payload']: # loop over to get each record
    try:
        raw=event['Records']['Payload']
        s=str(raw)[1:]
        
        records = ast.literal_eval(str(event['Records']['Payload'])[1:])
        print("-----------------")
        all_records.append(records)
    except:
        pass
t2=time.time()
print("time taken to loop: ", t2-t1) # prints 300 seconds (too much)

优化方案 #

• 替换低效的ast.literal_eval为json.loads
ast.literal_eval是为Python字面量设计的，处理JSON的效率远低于专门的json.loads。同时你之前用str(raw)[1:]截断字节流的方式错误且低效，应该直接解码字节流为字符串后解析：

  import json
  
  all_records = []
  t1 = time.time()
  for event in resp['Payload']:
      if 'Records' in event:
          raw_data = event['Records']['Payload'].decode('utf-8')
          # EventStream可能返回多行JSON，按行分割处理
          for line in raw_data.strip().split('\n'):
              if line:
                  record = json.loads(line)
                  all_records.append(record)
  t2 = time.time()
  print("time taken to loop: ", t2-t1)
  

• 移除循环内的打印操作
  原代码中print("-----------------")这类循环内的打印会大幅拖慢处理速度，尤其是数据量较大时，直接删除所有非必要的打印语句。

• 批量处理字节流，减少解析次数
  先收集所有Payload字节流，一次性解码解析，减少循环内的IO和解析开销：

  import json
  
  payload_data = b''
  for event in resp['Payload']:
      if 'Records' in event:
          payload_data += event['Records']['Payload']
  # 一次性解码并按行解析
  all_records = [json.loads(line) for line in payload_data.decode('utf-8').strip().split('\n') if line]
  

• 改用CSV输出格式替代JSON
  JSON序列化和反序列化的开销比CSV高，如果业务场景允许，修改S3 Select的输出配置为CSV，再用更快的CSV解析库处理：

  resp = s3_client.select_object_content(
      Bucket=bucket_name,
      Key=key,
      Expression=query,
      ExpressionType='SQL',
      InputSerialization={
          "CSV": {'FileHeaderInfo': 'USE'},
      },
      OutputSerialization={'CSV': {}}
  )
  
  # 使用csv模块解析
  import csv
  from io import StringIO
  
  payload_data = b''
  for event in resp['Payload']:
      if 'Records' in event:
          payload_data += event['Records']['Payload']
  csv_data = StringIO(payload_data.decode('utf-8'))
  reader = csv.DictReader(csv_data)
  all_records = list(reader)
  

• 升级boto3和botocore到最新版本
  旧版本的botocore在EventStream处理上可能存在性能瓶颈，执行以下命令升级：

  pip install --upgrade boto3 botocore
  

内容的提问来源于stack exchange，提问作者Asim