AWS Lambda读取S3中Parquet文件遇ArrowNotImplementedError求助

这个错误是因为当前使用的PyArrow版本未编译S3支持模块，无法直接读取S3 URI。以下是两种常用解决方式：

方法1：使用带S3支持的库 #

AWS Wrangler（awswrangler）是专为AWS服务优化的工具库，其依赖的PyArrow默认包含S3支持，且简化了S3与Parquet的交互逻辑。替换代码如下：

import awswrangler as wr
elif object_name.endswith(".parquet"):  
    rows = []
    if flow_name == "R50":
        path = f"s3://{bucket_name}/{object_prefix}"
        # 直接读取指定列并过滤非空值
        filtred_df = wr.s3.read_parquet(
            path,
            columns=["nom_archive", "nom_fichier"],
            filter=(wr.conditions.col("nom_archive").is_not_null() & wr.conditions.col("nom_fichier").is_not_null())
        )
        rows = filtred_df.to_dict(orient="records")
     df = pd.DataFrame(rows)

如果坚持使用原生PyArrow，需确保Lambda环境中的PyArrow包含S3支持：

• 通过Lambda Layer部署预编译的带S3支持的PyArrow包（推荐使用conda-forge构建的版本）
• 打包部署包时，使用pip install pyarrow[s3]安装包含S3依赖的版本（建议在Amazon Linux 2环境下构建，保证兼容性）

方法2：先下载文件到Lambda临时目录再读取 #

Lambda提供/tmp临时存储（可配置最大10GB），先将S3文件下载到本地，再用PyArrow读取本地文件，绕开S3支持依赖：

import boto3
import pyarrow.parquet as pq

s3_client = boto3.client('s3')

elif object_name.endswith(".parquet"):  
    rows = []
    if flow_name == "R50":
        # 下载文件到临时目录
        temp_file_path = f"/tmp/{object_name}"
        s3_client.download_file(bucket_name, f"{object_prefix}{object_name}", temp_file_path)
        
        # 读取指定列并过滤
        table = pq.read_table(temp_file_path, columns=["nom_archive", "nom_fichier"])
        filtered_table = table.filter(
            (table["nom_archive"].is_valid()) & (table["nom_fichier"].is_valid())
        )
        filtred_df = filtered_table.to_pandas()
        rows = filtred_df.to_dict(orient="records")
     df = pd.DataFrame(rows)

除了仅加载所需列，还有以下优化方向：

• 分批次读取：针对单文件过大的场景，使用PyArrow的iter_batches分批次处理数据，避免一次性加载全量数据到内存：

  reader = pq.ParquetFile(temp_file_path)
  for batch in reader.iter_batches(columns=["nom_archive", "nom_fichier"], batch_size=10000):
      filtered_batch = batch.filter(
          (batch["nom_archive"].is_valid()) & (batch["nom_fichier"].is_valid())
      )
      # 处理当前批次数据，比如追加到结果列表
      batch_rows = filtered_batch.to_pandas().to_dict(orient="records")
      rows.extend(batch_rows)
  

• 使用S3 Select：直接在S3端完成列过滤和行筛选，仅返回需要的数据，无需下载整个文件，大幅降低内存和带宽消耗：

  s3_client = boto3.client('s3')
  response = s3_client.select_object_content(
      Bucket=bucket_name,
      Key=f"{object_prefix}{object_name}",
      ExpressionType='SQL',
      Expression="SELECT nom_archive, nom_fichier FROM s3object s WHERE s.nom_archive IS NOT NULL AND s.nom_fichier IS NOT NULL",
      InputSerialization={'Parquet': {}},
      OutputSerialization={'JSON': {}}
  )
  # 解析返回的数据流
  for event in response['Payload']:
      if 'Records' in event:
          data = event['Records']['Payload'].decode('utf-8')
          # 按行解析JSON数据并追加到结果
          for line in data.strip().split('\n'):
              rows.append(json.loads(line))
  

• 提升Lambda资源配置：Lambda的内存与CPU、网络带宽正相关，调高内存配置（最大支持10GB）可提升处理速度，同时可配置更大的临时存储（最大10GB），满足大文件下载需求。

• 并行处理多文件：S3分区的10个文件可通过Lambda异步调用、Step Functions或SQS触发多个Lambda实例并行处理，避免单个实例负载过高。

• 使用Lambda Layer管理依赖：将PyArrow、awswrangler等依赖打包为Lambda Layer，减少部署包大小，同时实现依赖复用，简化更新流程。

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