PySpark读取S3上持续更新Parquet文件的技术疑问

1. 持续写入Parquet对PySpark访问/性能的影响及解决 #

• 潜在问题：
  • 读取到未完全写入的Parquet文件会直接抛出解析异常，导致Job失败；
  • CDC持续生成大量小Parquet文件，PySpark读取时需遍历大量对象，增加元数据加载时间，拉低处理性能。
• 解决方法：
  • 让CDC引擎采用原子写入：先将数据写入S3临时目录（如tmp/），文件写入完成后再重命名到目标目录，PySpark仅读取目标目录文件，避免接触未完成的临时文件；
  • 按时间/批次分区存储：比如按yyyy-MM-dd-HH或CDC批次ID分区，Glue Job只处理已完成的分区（可通过CDC批次状态标记或时间阈值过滤）；
  • Glue Job中开启小文件合并：读取后执行repartition(n)或使用Glue的mergeFiles参数，减少后续处理的文件数量；
  • 启用S3前缀列表：让PySpark只扫描指定前缀的文件，缩小扫描范围。

2. Parquet快照提取与自动化增量快照实现 #

• 能否提取快照：完全可以，快照本质是某一时间点的完整/增量数据集。
• 快照实现方式：
  • 基于时间戳的快照：利用CDC同步时保留的源库时间戳（如updated_at或CDC同步时间），在Glue Job中过滤updated_at <= 快照时间的数据，写入单独的快照目录（如snapshots/2024-05-20/）；
  • 基于S3版本控制的快照：开启S3桶的版本控制，通过指定时间点或版本ID，直接获取该时刻的Parquet文件集合作为快照。
• 自动化增量快照：
  • 用CloudWatch Events定时触发Glue Job（如每小时/每天），Job执行时自动获取上一次快照的时间戳，过滤增量数据生成新快照；
  • 结合Glue Workflow：编排CDC批次监控、数据校验、快照生成流程，当CDC完成一批数据同步后，自动触发增量快照生成；
  • 用Glue Data Catalog管理快照分区：将快照按时间分区注册到Catalog，后续转换操作直接读取对应分区即可。

3. Parquet格式损坏导致数据不一致的解决 #

• 常见原因：CDC写入时网络中断导致文件不完整、多进程并发写入同一Parquet文件、CDC引擎的Parquet序列化逻辑不规范。
• 修复与预防方案：
  • 强制CDC引擎使用原子写入流程（临时文件→重命名），杜绝半写入文件；
  • 检查CDC引擎的Parquet配置：确保使用兼容的Parquet版本，开启写入校验（如Parquet的checksum校验）；
  • Glue Job中添加文件校验逻辑：捕获Parquet读取异常，跳过损坏文件并记录日志，示例代码：

    from pyspark.sql.utils import AnalysisException
    
    try:
        df = spark.read.parquet("s3://your-bucket/cdc-data/")
    except AnalysisException as e:
        # 跳过损坏文件，指定坏记录存储路径
        df = spark.read.option("badRecordsPath", "s3://your-bucket/bad-records/").parquet("s3://your-bucket/cdc-data/")
    

  • 启用S3对象锁：设置文件写入后的不可修改期，防止中途被篡改；
  • 定期用Parquet工具做完整性校验：在Glue Job中调用parquet-tools check命令，扫描损坏文件并触发修复。

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