Azure Databricks中PySpark流式处理ADLS2数据增量更新问题

问题背景 #

使用Azure Databricks，数据存储于ADLS2，运行时版本10.4 LTS（可按需升级）。现有Pimproduct表结构如下：

字段说明：

• id：商品唯一ID
• name：商品名称
• action：行操作类型（A=新增，D=删除）
• dlk_last_modified：表中插入日期
• pim_timestamp：源系统提取日期

业务需求：每15分钟接收修改数据文件，按以下逻辑处理每个ID的最新pim_timestamp记录：

1. 若action=A且ID不存在，新增该行；
2. 若action=A且ID已存在，替换该ID的现有记录；
3. 若action=D，删除该ID的记录。

原每日批量处理代码：

from pyspark.sql import functions as F, Window as W

df = spark.table("Pimproduct").unionByNames(
    spark.read.format("avro").load("/path/to/daily/data")
)

df = df.withColumn(
    "rn",
    F.row_number().over(W.partitionBy("id").orderBy(F.col("pim_timestamp").desc())),
)

df = df.where("rn = 1").where("action <> 'D'")

df.write.saveAsTable("pimproduct", format="delta", mode="overwrite")

尝试转为流式处理时，使用以下代码报错：

import tempfile
from pyspark.sql import functions as F, Window as W

df = spark.readStream.table("Pimproduct").unionByNames(
    spark.readStream.schema(schema).format("avro").load("/path/to/daily/data")
)

df = df.withColumn(
    "rn",
    F.row_number().over(W.partitionBy("id").orderBy(F.col("pim_timestamp").desc())),
)

df = df.where("rn = 1").where("action <> 'D'")


with tempfile.TemporaryDirectory() as d:
    df.writeStream.toTable("Pimproduct", checkpointLocation=d)

报错信息：

AnalysisException: Non-time-based windows are not supported on streaming DataFrames/Datasets;

解决方案 #

流式处理中无法使用非时间窗口的row_number操作，正确方式是利用Delta Lake的merge操作实现新增、更新和删除逻辑，具体实现如下：

1. 读取流式Avro数据 #

先读取流数据，无需与现有表union，后续通过merge完成合并：

from pyspark.sql import functions as F

# 从现有表获取Schema，或手动定义
schema = spark.table("Pimproduct").schema

# 读取流式Avro数据
stream_df = spark.readStream \
    .schema(schema) \
    .format("avro") \
    .load("/path/to/streaming/data")

2. 用foreachBatch执行Merge逻辑 #

通过foreachBatch在每个微批中执行Delta Lake的merge操作，处理业务规则：

def merge_batch(batch_df, batch_id):
    # 为当前批次数据创建临时视图
    batch_df.createOrReplaceTempView("temp_updates")
    
    # 执行Merge操作，匹配ID并按action处理
    spark.sql("""
        MERGE INTO Pimproduct target
        USING temp_updates source
        ON target.id = source.id
        WHEN MATCHED AND source.action = 'A' AND source.pim_timestamp > target.pim_timestamp THEN
            UPDATE SET *
        WHEN NOT MATCHED AND source.action = 'A' THEN
            INSERT *
        WHEN MATCHED AND source.action = 'D' THEN
            DELETE
    """)

# 启动流处理
stream_query = stream_df.writeStream \
    .foreachBatch(merge_batch) \
    .option("checkpointLocation", "/dbfs/path/to/persistent/checkpoint")  # 必须用持久化路径，不能用临时目录
    .trigger(processingTime="15 minutes")  # 按数据接收频率设置触发间隔
    .start()

stream_query.awaitTermination()

关键说明 #

• checkpointLocation：必须使用DBFS等持久化路径，临时目录会导致流重启后丢失状态；
• Merge逻辑：
  • 匹配到相同ID且action为A、源数据时间戳更新时，执行全量更新；
  • 未匹配到ID且action为A时，插入新记录；
  • 匹配到ID且action为D时，删除对应记录；
• Trigger设置：按15分钟间隔触发，与数据接收频率对齐。

可选优化 #

若同一批次中存在同一ID的多条记录，可先在微批内去重，保留最新pim_timestamp记录：

def merge_batch(batch_df, batch_id):
    # 微批内去重，保留每个ID的最新记录
    deduplicated_df = batch_df.withColumn(
        "rn",
        F.row_number().over(F.window.partitionBy("id").orderBy(F.col("pim_timestamp").desc()))
    ).where("rn = 1").drop("rn")
    
    deduplicated_df.createOrReplaceTempView("temp_updates")
    
    # 执行Merge逻辑
    spark.sql("""
        MERGE INTO Pimproduct target
        USING temp_updates source
        ON target.id = source.id
        WHEN MATCHED AND source.action = 'A' THEN
            UPDATE SET *
        WHEN NOT MATCHED AND source.action = 'A' THEN
            INSERT *
        WHEN MATCHED AND source.action = 'D' THEN
            DELETE
    """)

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