Delta Streaming简单迁移任务耗时过长问题排查求助

一、代码中的明显问题 #

1. 读取端冗余的ignoreChanges配置 #

你的source_data函数中使用了option('ignoreChanges', True)，这个参数是为CDC（变更数据捕获）场景设计的，会强制Spark对Delta表执行全表扫描并追踪数据变更，带来大量额外计算开销。而你的场景是一次性全量迁移（trigger(once=True)），完全不需要这个配置，直接移除即可：

def source_data(schema: str, table: str, checkpoint: str):
    try:
      print('In getData_ae')
      df = spark.readStream.format('delta') \
        .option('startingOffsets', 'earliest') \
        .option('checkpointLocation', checkpoint) \
        .table(f'{schema}.{table}')
      return df
    except Exception as error:
        traceback.print_exc()
        sys.exit(1)

2. 冗余的foreachBatch与错误的写入模式 #

你用foreachBatch封装写入逻辑，但对于一次性全量迁移场景，foreachBatch完全多余——它会把数据流拆分成批次处理，增加调度和元数据操作的开销。更关键的是，dummy_write中使用mode='overwrite'，会导致每个批次都删除整个表的旧数据并重新写入元数据，对560万行的表来说，这会产生巨量IO和元数据操作，直接拖慢速度。

替换为直接用writeStream的toTable方法，使用append模式（空表场景下append与overwrite效果一致，但避免了全表删除的开销）：

df.writeStream.format("delta") \
      .option("checkpointLocation", 's3://some_location') \
      .option("path", 's3://some_s3_path') \
      .trigger(once=True) \
      .toTable('some_schema.some_table') \
      .awaitTermination()

二、Delta表与Spark配置优化 #

1. 开启Delta自动优化 #

写入前开启Delta自动优化配置，让系统自动合并小文件、优化写入路径，减少S3上的IO开销：

spark.conf.set("spark.databricks.delta.autoOptimize.optimizeWrite", "true")
spark.conf.set("spark.databricks.delta.autoOptimize.autoCompact", "true")

2. 调整Spark并行度 #

你的集群有128个Worker核，默认的spark.sql.shuffle.partitions为200，可调整为核数的2倍（如256），让任务充分利用集群资源：

spark.conf.set("spark.sql.shuffle.partitions", "256")

也可提前通过repartition调整数据分区数，避免后续写入时的分区不均衡：

df = source_data('schema', 'table', 'checkpoint').repartition(256)

3. 清理Checkpoint目录 #

确保checkpointLocation指向的S3目录是全新无残留数据的。如果之前有失败任务的checkpoint文件残留，可能导致Spark重复处理旧数据或陷入状态恢复死循环。

三、其他排查方向 #

• 优化原表存储：如果landing_table存在大量小文件，先对原表执行一次OPTIMIZE操作减少IO开销：

  OPTIMIZE schema.table ZORDER BY (your_key_column)
  

• 排查数据倾斜：查看Application Master UI中卡住阶段的任务数据量，如果存在单个任务处理数据量远大于其他任务的情况，可通过加盐分区或调整分区数解决。
• 检查S3网络：确认集群到S3的网络带宽充足，没有限流或延迟过高的情况。

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