关于Databricks中foreachBatch实现流式去重的三类技术疑问

以下是Databricks官方示例中ADE 3.1 - Streaming Deduplication章节的流式去重代码：

from pyspark.sql import functions as F
     
json_schema = "device_id LONG, time TIMESTAMP, heartrate DOUBLE"
     
deduped_df = (spark.readStream
              .table("bronze")
              .filter("topic = 'bpm'")
              .select(F.from_json(F.col("value").cast("string"), json_schema).alias("v"))
              .select("v.*")
              .withWatermark("time", "30 seconds")
              .dropDuplicates(["device_id", "time"]))
     
sql_query = """
     MERGE INTO heart_rate_silver a
     USING stream_updates b
     ON a.device_id=b.device_id AND a.time=b.time
     WHEN NOT MATCHED THEN INSERT *
    """
     
class Upsert:
  def __init__(self, sql_query, update_temp="stream_updates"):
    self.sql_query = sql_query
    self.update_temp = update_temp 
     
  def upsert_to_delta(self, microBatchDF, batch):
        microBatchDF.createOrReplaceTempView(self.update_temp)
        microBatchDF._jdf.sparkSession().sql(self.sql_query)
     
streaming_merge = Upsert(sql_query)
query = (deduped_df.writeStream
   .foreachBatch(streaming_merge.upsert_to_delta)  # run query for each batch
   .outputMode("update")
   .option("checkpointLocation", f"{DA.paths.checkpoints}/recordings")
   .trigger(availableNow=True)
   .start())
     
query.awaitTermination()

疑问解答 #

1. 为何要定义Upsert类并使用foreachBatch方法？ #

• Spark Structured Streaming原生的writeStream输出模式（如append、update、complete）无法直接对Delta表执行MERGE操作。foreachBatch允许我们在每个微批处理的上下文里，自定义执行逻辑——这里就是用MERGE实现仅插入目标表中不存在的记录，避免重复写入。
• 定义Upsert类是为了封装MERGE的SQL逻辑和临时表名称，让代码更模块化、可复用。如果直接写匿名函数，逻辑分散，后续修改维护麻烦，封装成类后可方便调整SQL或临时表参数。

2. 若不使用foreachBatch，仅通过dropDuplicates(["device_id", "time"])是否能确保无重复记录？ #

• 不能完全确保。dropDuplicates结合withWatermark只能在当前流处理的30秒窗口内去重，仅保证窗口内不会有重复的device_id+time记录。但如果流处理重启、有晚到的记录（超过30秒水印窗口），或者目标表中已存在历史重复记录，单纯的流式去重无法覆盖这些场景。
• 另外，原生append模式写入Delta表时，即使流内去重了，若上游重发重复数据，仍会写入目标表；而MERGE操作会在写入前检查目标表是否已有该记录，从根源避免重复。

3. Upsert类的upsert_to_delta方法需要microBatchDF和batch两个参数，但调用时未传递，参数值如何获取？ #

• 这是Spark Structured Streaming的foreachBatch机制自动传递的参数。当你把方法传递给foreachBatch时，Spark会在每个微批触发时，自动将当前微批的DataFrame（microBatchDF）和批处理ID（batch）作为参数传入该方法，无需手动传递。
• 只需保证传递的方法签名符合要求（接收微批DF和批ID两个参数），Spark会负责注入参数值。

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