基于日期参数的数据重处理方案咨询（Azure Databricks Notebook）

问题背景 #

• 数据架构：ADLS Gen2作为存储底座，Landing Zone通过ADF复制活动从SQL Server拉取源数据；Bronze层由Databricks PySpark脚本转为Delta表；Silver、Gold层依赖Databricks Python代码完成清洗聚合
• 当前运行逻辑：每日增量加载，Landing Zone保留7天归档数据，Bronze层采用每日truncate and load模式
• 核心需求：传入指定日期参数后，触发全链路重处理——读取Landing Zone对应日期的历史文件，重新执行Bronze→Silver→Gold的完整流程
• 此前尝试：通过ADF或Databricks Job调度Notebook，用dbutils.fs.ls按日期枚举文件但未成功落地

解决方案一：Azure Data Factory（ADF）驱动的参数化重处理 #

1. 配置管道参数 #

• 在ADF中创建新管道，添加字符串类型参数reprocessDate，格式与Landing Zone的日期路径/命名规则一致（例如yyyy-MM-dd）
• 确保该参数能传递到管道内所有关联活动

2. 前置校验：确认目标文件存在 #

• 添加Get Metadata活动，指向Landing Zone的日期路径：@concat('abfss://landing@{your-adls-account}.dfs.core.windows.net/', pipeline().parameters.reprocessDate)
• 勾选Exists作为字段输出，后续通过If Condition活动判断文件是否存在，避免空跑流程

3. 触发Databricks Notebook并传递参数 #

• 添加Databricks Notebook活动，关联目标工作区与主处理Notebook
• 在Base Parameters中添加键值对：reprocess_date → @pipeline().parameters.reprocessDate

4. 修改Notebook代码适配参数 #

替换原有的自动取当日逻辑，改为读取传入参数，核心代码示例：

# 读取重处理日期参数，兼容日常自动运行（默认取当日）
from datetime import date
try:
    reprocess_date = dbutils.widgets.get("reprocess_date")
except:
    reprocess_date = str(date.today())

# 构建Landing Zone目标路径
landing_base_path = "abfss://landing@{your-adls-account}.dfs.core.windows.net/"
target_landing_path = f"{landing_base_path}{reprocess_date}/"

# 校验路径下存在文件
file_list = dbutils.fs.ls(target_landing_path)
if not file_list:
    raise RuntimeError(f"No source files found for date: {reprocess_date}")

# 读取源文件（替换为你的实际格式：csv/parquet等）
source_df = spark.read.format("csv").option("header", "true").load(target_landing_path)

# Bronze层：Truncate后写入Delta表
bronze_table_path = "abfss://bronze@{your-adls-account}.dfs.core.windows.net/{your-table-name}/"
spark.sql(f"TRUNCATE TABLE delta.`{bronze_table_path}`")
source_df.write.format("delta").mode("append").save(bronze_table_path)

# 调用Silver、Gold层Notebook并传递日期参数
dbutils.notebook.run("/path/to/silver-processing-notebook", 3600, {"reprocess_date": reprocess_date})
dbutils.notebook.run("/path/to/gold-processing-notebook", 3600, {"reprocess_date": reprocess_date})

解决方案二：Databricks Job直接驱动的重处理 #

1. 创建带参数的Databricks Job #

• 在Databricks UI中新建Job，选择主处理Notebook
• 在Parameters模块添加reprocess_date参数，设置默认值为{{current_date()}}（兼容日常自动调度）

2. Notebook代码适配 #

与ADF方案中的Notebook代码完全一致，核心是通过dbutils.widgets读取参数并校验文件存在

3. 触发重处理 #

• 手动触发：进入Job页面点击Run now，输入指定日期参数
• 自动化触发：调用Databricks Jobs API，传入reprocess_date参数实现批量重处理

常见问题排查 #

• 路径匹配失败：确保Landing Zone的文件路径/命名与参数格式完全对齐（例如参数是2024-05-20，文件需存放在landing/2024-05-20/下）
• 参数传递失效：检查ADF活动的参数映射是否正确，或Databricks Job的参数是否在Notebook中通过dbutils.widgets.get()正确读取
• 权限问题：确认Databricks服务主体拥有Landing Zone的读取权限，以及Bronze/Silver/Gold层的读写权限

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