Databricks中Spark Structured Streaming foreachBatch分区报错的替代方案咨询

你遇到的这个报错是因为Spark Structured Streaming的foreachBatch机制和直接写流的partitionBy选项无法兼容——partitionBy是针对内置流写入器的配置，而foreachBatch允许你自定义批处理逻辑，这时候需要把分区逻辑整合到你的自定义函数里。

下面提供两种可靠的替代方案，都能实现分区写入的同时保留foreachBatch的Upsert逻辑：

方案一：预先创建带分区的Delta表（推荐） #

Delta Lake本身支持分区表，我们可以先手动创建好带分区的目标表，之后在Upsert逻辑中，Delta会自动将数据写入对应的分区目录，不需要在writeStream中指定partitionBy。

步骤1：创建分区Delta表 #

先执行SQL创建目标表，指定year、month、day作为分区列：

CREATE TABLE silver (
  smtUidNr STRING,
  dcl STRING,
  inv STRING,
  evt STRING,
  smt STRING,
  msgTs TIMESTAMP,
  msgInfSrcCd STRING,
  year INT,
  month INT,
  day INT
)
PARTITIONED BY (year, month, day)
USING DELTA
LOCATION 'abfss://dump@mcfdatalake.dfs.core.windows.net/main_data/'

如果你的表已经存在，可以用ALTER TABLE添加分区列：

ALTER TABLE silver ADD COLUMNS (year INT, month INT, day INT);
ALTER TABLE silver PARTITIONED BY (year, month, day);

步骤2：修改Upsert函数，包含分区列 #

确保你的微批数据中包含分区列（可以从msgTs提取），然后在MERGE语句中包含这些列：

def upsertToDelta(microBatchOutputDF: DataFrame, batchId: Long) {
  // 从msgTs提取年月日作为分区列
  val dfWithPartitions = microBatchOutputDF
    .withColumn("year", year(col("msgTs")))
    .withColumn("month", month(col("msgTs")))
    .withColumn("day", dayofmonth(col("msgTs")))
  
  dfWithPartitions.createOrReplaceTempView("updates")
  
  dfWithPartitions.sparkSession.sql(s"""
    MERGE INTO silver as r
    USING (
      SELECT smtUidNr, dcl, inv, evt, smt, msgTs, msgInfSrcCd, year, month, day
      FROM (
        SELECT 
          smtUidNr, msgTs, dcl, inv, evt, smt, msgInfSrcCd, year, month, day,
          RANK() OVER (PARTITION BY smtUidNr ORDER BY msgTs DESC) as rank,
          ROW_NUMBER() OVER (PARTITION BY smtUidNr ORDER BY msgTs DESC) as row_num
        FROM updates
      )
      WHERE rank = 1 AND row_num = 1
    ) as u
    ON u.smtUidNr = r.smtUidNr
    WHEN MATCHED AND u.msgTs > r.msgTs THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
  """)
}

步骤3：启动流任务（移除partitionBy） #

splitDF.writeStream
  .format("delta")
  .foreachBatch(upsertToDelta _)
  .outputMode("append")
  .option("checkpointLocation", "abfss://checkpoint@mcfdatalake.dfs.core.windows.net/kjd/test/")
  .start("abfss://dump@mcfdatalake.dfs.core.windows.net/main_data/")

方案二：在foreachBatch函数中手动处理分区写入 #

如果不想预先创建表，也可以在微批函数中直接对DataFrame进行分区写入，但这种方式需要确保每次写入都符合分区逻辑，且性能略逊于方案一（因为Delta无法提前优化分区布局）。

def upsertToDelta(microBatchOutputDF: DataFrame, batchId: Long) {
  val dfWithPartitions = microBatchOutputDF
    .withColumn("year", year(col("msgTs")))
    .withColumn("month", month(col("msgTs")))
    .withColumn("day", dayofmonth(col("msgTs")))
  
  // 先获取最新版本的数据（去重逻辑）
  val latestUpdates = dfWithPartitions
    .withColumn("rank", rank().over(Window.partitionBy("smtUidNr").orderBy(col("msgTs").desc)))
    .withColumn("row_num", row_number().over(Window.partitionBy("smtUidNr").orderBy(col("msgTs").desc)))
    .filter("rank = 1 AND row_num = 1")
  
  // 写入时指定分区列
  latestUpdates.write
    .format("delta")
    .mode("append")
    .partitionBy("year", "month", "day")
    .save("abfss://dump@mcfdatalake.dfs.core.windows.net/main_data/")
  
  // 注意：这种方式没有MERGE逻辑，如果你需要Upsert，还是得用方案一的MERGE语句
}

关键注意事项 #

1. 无论哪种方案，都需要确保你的数据中存在分区列（year/month/day），可以通过时间戳字段动态提取。
2. 方案一的MERGE操作是Delta Lake的核心特性，能确保数据的幂等性和正确性，是生产环境的首选。
3. 如果你使用的是Databricks Runtime，也可以考虑使用autoOptimize和autoCompact选项来优化分区表的性能。

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