Databricks批式readStream作业多源匹配同一目标行报错问题

问题场景 #

• Bronze层通过spark.readStream读取Delta表增量数据，使用trigger(availableNow=True)每日以批式模式运行，而非持续流
• Silver层作业失败后，重新执行会同时处理新数据与上一批次未成功处理的数据，若存在重复主键，会触发「多源匹配并尝试修改同一目标行」的MERGE报错
• 曾尝试两种方案但受限于流API特性失败：
  • 直接对流DF按增量字段排序后分组聚合：spark.readStream不支持orderBy操作
  • 用无时间窗口的窗口函数筛选最新数据：流处理不允许定义无时间边界的窗口

当前代码采用分组聚合仍有问题，核心误区是误将流DataFrame当作静态DataFrame处理，原代码如下：

columnskeys_list = columnskeys_str.split(",")
        
bronzeDF= (spark.readStream
      .format("delta")
      .table(source_table))
      #.orderBy(col(incr_column).desc())
      
#window_spec = Window.partitionBy(*columnskeys_list).orderBy(desc(incr_column))

#bronzeDF = bronzeDF.withColumn("rn", row_number().over(window_spec))

#bronzeDF = bronzeDF.filter("rn = 1")
aggregation_columns = [first(col(c)).alias(c) for c in bronzeDF.columns if c not in columnskeys_list and c != incr_column ]
aggregation_columns.append(max(col(incr_column)).alias(incr_column))
group_columns = [col(col_name) for col_name in columnskeys_list]
bronzeDF = bronzeDF.groupBy(*group_columns).agg(*aggregation_columns)


def upsertToDelta(microBatchOutputDF, batchId, target_table, condition):
  microBatchOutputDF.createOrReplaceTempView("updates")

  microBatchOutputDF.sparkSession.sql(f"""
    MERGE INTO {target_table}
    USING updates
    ON {condition}
    WHEN MATCHED
      THEN UPDATE SET *
    WHEN NOT MATCHED
      THEN INSERT *
  """)

def bronzeToSilver(target_table, condition, checkpoint):
  if spark.catalog.tableExists(target_table):
    (
      bronzeDF
      .na.fill(0).na.fill('')
      .writeStream
      .format("delta")
      .foreachBatch(lambda df, batchId: upsertToDelta(df, batchId, target_table, condition))
      .trigger(availableNow=True)
      .outputMode("complete")
      .option("checkpointLocation", checkpoint)
      .start().awaitTermination()
    )
  else:
    (
      bronzeDF
      .na.fill(0).na.fill('')
      .writeStream.format("delta")
      .outputMode("complete")
      .trigger(availableNow=True)
      .option("checkpointLocation", checkpoint)
      .toTable(target_table)
      .awaitTermination()
    )

解决方案 #

核心思路 #

流处理API对排序、无时间窗口的操作有严格限制，但每个微批对应的DataFrame是静态的，因此可以将去重逻辑移到foreachBatch内部处理，在每个微批中筛选出每个主键的最新记录后再执行MERGE。

修正后的代码 #

from pyspark.sql.window import Window
from pyspark.sql.functions import col, row_number, desc

columnskeys_list = columnskeys_str.split(",")
        
# 流处理仅负责读取增量数据，不做去重/聚合
bronzeDF= (spark.readStream
      .format("delta")
      .table(source_table))


def upsertToDelta(microBatchOutputDF, batchId, target_table, condition, incr_column):
  # 在微批内部处理静态DF，用窗口函数筛选每个主键的最新记录
  window_spec = Window.partitionBy(*columnskeys_list).orderBy(desc(incr_column))
  deduplicated_df = microBatchOutputDF.withColumn("rn", row_number().over(window_spec)) \
                                     .filter("rn = 1") \
                                     .drop("rn")
  
  deduplicated_df.createOrReplaceTempView("updates")

  deduplicated_df.sparkSession.sql(f"""
    MERGE INTO {target_table}
    USING updates
    ON {condition}
    WHEN MATCHED
      THEN UPDATE SET *
    WHEN NOT MATCHED
      THEN INSERT *
  """)

def bronzeToSilver(target_table, condition, checkpoint, incr_column):
  if spark.catalog.tableExists(target_table):
    (
      bronzeDF
      .na.fill(0).na.fill('')
      .writeStream
      .format("delta")
      .foreachBatch(lambda df, batchId: upsertToDelta(df, batchId, target_table, condition, incr_column))
      .trigger(availableNow=True)
      .outputMode("append")  # 改用append模式，因为去重后每个微批输出唯一主键记录
      .option("checkpointLocation", checkpoint)
      .start().awaitTermination()
    )
  else:
    # 首次初始化表时，先做全量去重
    initial_df = spark.read.format("delta").table(source_table)
    window_spec = Window.partitionBy(*columnskeys_list).orderBy(desc(incr_column))
    deduplicated_initial_df = initial_df.withColumn("rn", row_number().over(window_spec)) \
                                        .filter("rn = 1") \
                                        .drop("rn") \
                                        .na.fill(0).na.fill('')
    
    deduplicated_initial_df.write.format("delta").saveAsTable(target_table)

关键优化点 #

1. 去重逻辑下移到微批内部：利用微批DF的静态特性，使用窗口函数安全筛选每个主键的最新记录，避开流API的限制
2. 修改输出模式为append：去重后每个微批输出的是唯一主键记录，无需用complete模式，减少不必要的全量计算
3. 首次初始化表时单独处理：避免流处理首次运行时的全量重复问题，直接对静态表做全量去重后写入

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