如何使用Azure Databricks Delta Lake链接服务实现Delta表增改及审计方案

一、ADF侧适用的活动组件 #

针对Delta表的插入/更新操作，ADF优先选用以下两种活动替代原SQL存储过程方案：

• Databricks Notebook活动：灵活性最高，适配复杂审计逻辑。提前在Databricks编写好处理Delta表增改的Notebook，ADF通过该活动调用Notebook并传入管道详情参数（如管道ID、执行状态、触发时间等）。
• Databricks Spark SQL活动：适合简单插入场景。直接在活动中编写Spark SQL语句（如INSERT INTO audit_logs VALUES (?,?,?)），绑定ADF系统变量（如@pipeline().PipelineId、@pipeline().TriggerTime）作为参数，无需额外编写Notebook。

注意：Copy Data活动不适合该场景，它主打批量数据迁移，无法支持带动态参数的单条/少量审计日志写入。

二、ADF与Databricks Notebook共享Delta表的最佳实践 #

1. 统一表访问方式 #

提前在Databricks创建外部Delta表，指定存储路径（如ADLS Gen2路径），确保ADF和Notebook都通过该表或直接访问存储路径操作同一份数据：

CREATE TABLE IF NOT EXISTS audit_logs (
    pipeline_id STRING COMMENT 'ADF管道ID',
    notebook_path STRING COMMENT 'Databricks Notebook路径',
    operation_type STRING COMMENT '操作类型：pipeline_execution/api_call',
    status STRING COMMENT '执行状态：success/failed/running',
    execution_time TIMESTAMP COMMENT '执行时间',
    details STRING COMMENT '详细信息'
) USING DELTA 
LOCATION 'abfss://<container-name>@<storage-account>.dfs.core.windows.net/audit/logs/path';

2. 规范参数传递与数据写入 #

• ADF侧写入：通过Databricks活动的参数传递功能，将管道系统变量传给Notebook，Notebook中通过dbutils.widgets.get()获取参数后写入Delta表：
  示例Notebook代码片段：

  from pyspark.sql import Row
  from datetime import datetime
  
  # 获取ADF传入的参数
  pipeline_id = dbutils.widgets.get("pipeline_id")
  status = dbutils.widgets.get("status")
  details = dbutils.widgets.get("details")
  
  # 构造审计日志行
  audit_row = Row(
      pipeline_id=pipeline_id,
      notebook_path="",
      operation_type="pipeline_execution",
      status=status,
      execution_time=datetime.now(),
      details=details
  )
  
  # 写入Delta表
  spark.createDataFrame([audit_row]).write.mode("append").saveAsTable("audit_logs")
  

• Databricks Notebook侧写入：调用API后，直接用Spark SQL或PySpark将API执行状态写入同一张Delta表：

  import requests
  from pyspark.sql import Row
  from datetime import datetime
  
  # 调用API逻辑
  api_response = requests.get("https://your-api-endpoint.com")
  api_status = "success" if api_response.status_code == 200 else "failed"
  api_details = f"API响应码：{api_response.status_code}"
  
  # 构造审计日志行
  audit_row = Row(
      pipeline_id="",
      notebook_path=dbutils.notebook.entry_point.getDbutils().notebook().getContext().notebookPath().get(),
      operation_type="api_call",
      status=api_status,
      execution_time=datetime.now(),
      details=api_details
  )
  
  # 写入Delta表
  spark.createDataFrame([audit_row]).write.mode("append").saveAsTable("audit_logs")
  

3. 并发与事务控制 #

Delta Lake原生支持ACID事务，ADF和Notebook同时写入不会出现数据冲突：

• 插入操作统一使用append模式，避免覆盖现有数据；
• 若需更新已有日志（如补全管道最终状态），使用MERGE INTO语法实现原子更新：

  MERGE INTO audit_logs t
  USING (SELECT ? AS pipeline_id, ? AS status) s
  ON t.pipeline_id = s.pipeline_id
  WHEN MATCHED THEN UPDATE SET t.status = s.status;
  

4. 权限与访问控制 #

• 确保ADF链接服务使用的服务主体，以及Databricks集群使用的服务主体，都拥有Delta表存储路径的Storage Blob Data Contributor权限；
• 在Databricks中为相关用户/服务主体配置audit_logs表的读写权限。

5. 日志字段统一 #

统一ADF管道日志和Notebook API日志的字段结构，便于后续日志查询、分析和可视化。

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