Databricks每日触发数据管道异步JSON数据漏采问题求解

针对你遇到的问题，这里提供几个可行的解决方案，按落地复杂度和灵活性排序：

1. 基于文件修改时间的增量处理逻辑 #

不要仅按日期标记文件"已处理"，而是跟踪文件的最后修改时间戳，每次管道触发时对比文件最新修改时间与上次处理的时间戳，判断是否需要重新读取并合并数据：

• 用一个元数据Delta表来存储每个日期对应的文件最后处理时间戳
• 每次执行时，先获取当日JSON文件的当前修改时间，与元数据表记录的时间对比，若文件有更新则重新处理
• 处理完成后更新元数据表的时间戳，确保后续只处理新增的内容

示例代码片段：

from datetime import datetime

# 定义路径和日期
current_date = datetime.today().strftime("%Y-%m-%d")
json_file_path = f"/mnt/storage/{current_date}.json"
metadata_table_path = "/mnt/storage/metadata/processed_files"

# 获取文件当前修改时间
file_info = dbutils.fs.ls(json_file_path)[0]
current_modified_ts = file_info.modificationTime

# 读取元数据表中的上次处理记录
metadata_df = spark.read.format("delta").load(metadata_table_path)
last_processed_ts = None
if metadata_df.filter(f"date = '{current_date}'").count() > 0:
    last_processed_ts = metadata_df.filter(f"date = '{current_date}'").select("last_modified_ts").first()[0]

# 判断是否需要处理
if last_processed_ts is None or current_modified_ts > last_processed_ts:
    # 读取JSON文件
    raw_df = spark.read.json(json_file_path)
    # 增量合并到目标Delta表（假设按主键去重）
    raw_df.write.format("delta").mode("merge")\
        .option("mergeSchema", "true")\
        .option("mergeCondition", "target.id = source.id")\
        .saveAsTable("your_target_delta_table")
    
    # 更新元数据表
    update_df = spark.createDataFrame([(current_date, current_modified_ts)], ["date", "last_modified_ts"])
    update_df.write.format("delta").mode("overwrite")\
        .option("replaceWhere", f"date = '{current_date}'")\
        .save(metadata_table_path)

2. 拆分写入策略，避免单文件追加 #

协调数据写入端，每次生成新的JSON文件而非追加到同一日期文件，比如按批次时间戳命名（如2024-05-20_1430.json）：

• 管道只需跟踪已处理的文件列表（同样可以用元数据表存储已处理文件名）
• 每日触发时，扫描当日所有JSON文件，过滤掉已处理的，只处理新增文件
• 这种方式避免了同一文件的重复读取问题，逻辑更简单，适合写入端可调整的场景

3. 延迟管道触发时间 #

如果业务场景允许，将每日管道的触发时间调整到当日数据收集完全结束之后（比如凌晨2点）：

• 这是最简单的方案，但依赖业务数据的写入周期固定，无法应对突发的延迟写入场景

4. 改用Structured Streaming流处理 #

将批处理改为Structured Streaming，监听文件存储路径的变化，自动处理文件的新增或更新：

• 流处理可以准实时地捕获文件的追加内容，不需要手动管理处理状态
• 针对同一文件的多次追加，需要确保流处理能识别文件更新（部分存储系统需配置文件通知，或开启fileCreationTime=latest参数）

示例流处理代码：

stream_df = spark.readStream.format("json")\
    .option("fileCreationTime", "latest")\
    .load("/mnt/storage/json_files")

query = stream_df.writeStream.format("delta")\
    .option("checkpointLocation", "/mnt/storage/checkpoint")\
    .option("mergeSchema", "true")\
    .table("your_target_delta_table")

query.awaitTermination()

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