Spark Structured Streaming写入Delta表实现upsert/delete的最佳方案

生产环境最稳定、性能最好的实现方式是Delta Lake原生foreachBatch + merge语法，不需要自己维护流状态，所有增删改逻辑下沉到Delta的事务性merge能力实现，具体落地步骤如下：

• 流读取与预处理环节
  从Azure Event Hub读取消息流后，先完成既定的数据质量校验，过滤无主键、字段格式非法的脏数据，将二进制消息体解析为结构化DataFrame。务必保留两个核心字段：业务主键、操作类型字段（op，取值为created/updated/deleted），以及事件时间字段。建议基于事件时间设置watermark，避免流状态无限膨胀，允许晚到的阈值根据业务SLA配置即可，核心代码示例（PySpark/Scala逻辑一致）：

  from pyspark.sql import functions as F
  from delta.tables import DeltaTable
  
  # 读取Event Hub流，eh_conf为Event Hub连接配置
  stream_df = spark.readStream.format("eventhubs").options(**eh_conf).load()
  
  # 解析消息、执行数据质量规则
  parsed_df = stream_df.select(
      F.from_json(F.col("body").cast("string"), schema=biz_event_schema).alias("data")
  ).select("data.*")
   .filter(F.col("biz_id").isNotNull()) # 主键非空等自定义校验规则
   .withWatermark("event_time", "2 hours") # 按实际业务对晚到数据的容忍度调整
  

• 提前初始化目标Delta表
  首次任务启动前先创建目标Delta表，表结构与流字段对齐，若使用Delta 2.0及以上版本可显式指定业务主键，优化后续merge性能：

  CREATE TABLE IF NOT EXISTS dwd.biz_core_table (
      biz_id STRING,
      -- 其余业务字段
      op STRING,
      event_time TIMESTAMP,
      dt STRING
  ) USING DELTA
  PARTITIONED BY (dt)
  TBLPROPERTIES ('delta.primaryKey' = 'biz_id')
  

• 核心merge逻辑实现
  在foreachBatch算子中实现每个微批的写入逻辑，注意每个微批内先做去重：由于Event Hub存在至少一次投递语义，且事件可能乱序到达，同一个业务主键在单批次内可能存在多条操作记录，仅保留事件时间最新的一条即可，避免旧操作覆盖新数据。之后按操作类型执行merge：

  1. 目标表匹配到对应主键，且当前操作类型为deleted时，删除目标表记录
  2. 目标表匹配到对应主键，且当前操作类型为created/updated时，更新该行所有字段
  3. 目标表未匹配到对应主键，且当前操作类型为created/updated时，插入新行
     未匹配到主键的deleted操作直接跳过，避免无效写入。核心代码：

  def batch_merge(micro_batch_df, batch_id):
      # 单微批内按主键去重，仅保留最新事件
      deduped_df = micro_batch_df.dropDuplicatesWithinWatermark(
          ["biz_id"], 
          F.col("event_time").desc()
      )
      
      # 加载目标Delta表
      target_tbl = DeltaTable.forName(spark, "dwd.biz_core_table")
      
      # 执行事务性merge
      target_tbl.alias("tgt").merge(
          source = deduped_df.alias("src"),
          condition = "tgt.biz_id = src.biz_id AND tgt.dt = src.dt" # 加分区条件裁剪数据，大幅提升merge性能
      ).whenMatchedDelete(condition = "src.op = 'deleted'")
      .whenMatchedUpdateAll(condition = "src.op IN ('created', 'updated')")
      .whenNotMatchedInsertAll(condition = "src.op IN ('created', 'updated')")
      .execute()
  

• 启动流式任务
  配置检查点路径与触发间隔，启动流任务即可：

  stream_query = parsed_df.writeStream \
    .foreachBatch(batch_merge) \
    .option("checkpointLocation", "abfss://checkpoints@adlsaccount/biz_core_table/") \
    .trigger(processingTime="1 minute") # 按延迟要求调整批次间隔，也可配置availableNow=True做批流一体调度
    .start()
  
  stream_query.awaitTermination()
  

生产环境注意事项 #

• 不要尝试用mapGroupsWithState/flatMapGroupsWithState自行维护键值状态实现增删改，数据量上涨后会出现checkpoint膨胀、故障恢复耗时极长的问题，Delta merge方案将数据状态存储在Delta表本身，无额外状态维护成本，稳定性经过大量生产场景验证。
• merge条件中一定要加入分区字段的匹配规则，可将merge时扫描的文件量降低1-2个数量级，避免全表扫描带来的性能问题。
• 若后续需要将表的增删改变更同步到下游系统，可提前开启Delta的Change Data Feed特性，直接读取变更日志即可，不需要额外解析commit记录。
• 检查点目录必须存储在高可靠持久化存储上（如ADLS Gen2），任务运行期间不要随意删除检查点目录，否则会导致重复消费。

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