基于Delta Live Tables实现按检出分支配置不同存储路径

以下是几种实现根据当前代码分支调整DLT流水线存储位置的实用方案：

方案1：通过CI/CD结合Databricks CLI动态更新流水线配置 #

这种方式适合在分支触发CI/CD流程时，直接修改DLT流水线的存储路径参数：

• 提前准备DLT流水线的基础配置模板（如dlt_pipeline_template.json），将存储位置设为占位符：

  {
    "name": "my-dlt-pipeline",
    "storage": "${STORAGE_LOCATION}",
    "configuration": {},
    "clusters": [...],
    "libraries": [...]
  }
  

• 在CI/CD脚本中，根据当前分支设置对应的存储路径环境变量：

  # 示例：根据分支名判断存储路径
  case "$GIT_BRANCH" in
    feature/*)
      export STORAGE_LOCATION="dbfs:/pipelines/feature/my_dlt_pipeline"
      ;;
    development)
      export STORAGE_LOCATION="dbfs:/pipelines/dev/my_dlt_pipeline"
      ;;
    *)
      export STORAGE_LOCATION="dbfs:/pipelines/default/my_dlt_pipeline"
      ;;
  esac
  

• 使用sed替换模板中的占位符，再通过Databricks CLI更新流水线：

  sed "s/\${STORAGE_LOCATION}/${STORAGE_LOCATION}/" dlt_pipeline_template.json > dlt_pipeline_config.json
  databricks pipelines update --json-file dlt_pipeline_config.json
  

方案2：在DLT代码中通过参数化动态生成存储路径 #

如果需要更细粒度的表级存储控制，可以在DLT代码中引入分支参数，动态生成存储路径：

1. 在DLT流水线配置中添加自定义参数（如branch_name），部署时根据当前分支传入对应值
2. 在DLT代码中读取该参数并拼接存储路径：
   • Python示例：

     from dlt import config
     
     # 获取流水线传入的分支参数
     branch = config.get("branch_name", "default")
     base_storage = f"dbfs:/pipelines/{branch}/my_dlt_pipeline"
     
     @dlt.table(
         name="raw_user_data",
         storage_location=f"{base_storage}/raw/user_data"
     )
     def raw_user_data():
         return spark.read.csv("/path/to/raw/user_data.csv")
     

   • SQL示例：

     CREATE OR REFRESH STREAMING LIVE TABLE raw_user_data
     LOCATION "${branch_name}/raw/user_data"
     AS SELECT * FROM cloud_files("/path/to/raw/user_data", "csv");
     

方案3：直接通过Databricks CLI命令行指定存储路径 #

如果流水线ID固定，可以在CI脚本中直接根据分支调用CLI命令更新存储位置：

PIPELINE_ID="your-pipeline-uuid-here"

if [[ "$GIT_BRANCH" == "feature/"* ]]; then
  databricks pipelines update --pipeline-id $PIPELINE_ID --storage "dbfs:/pipelines/feature/my_dlt_pipeline"
elif [[ "$GIT_BRANCH" == "development" ]]; then
  databricks pipelines update --pipeline-id $PIPELINE_ID --storage "dbfs:/pipelines/dev/my_dlt_pipeline"
fi

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