如何通过Databricks资产包的目标环境确定表写入的Schema

方案1：通过Bundle配置传递命令行参数 #

直接修改你的bundle.yml，给不同环境定义变量，再把变量传给Python脚本：

1. 调整bundle.yml内容：

bundle:
  name: your-bundle-name

environments:
  dev:
    variables:
      target_env: "dev"
  prod:
    variables:
      target_env: "prod"

jobs:
  your-job-name:
    tasks:
      - task_key: write-to-schema
        spark_python_task:
          python_file: "/path/to/your/script.py"
          parameters: ["--env", "{{ bundle.target_env }}"]

2. 在Python脚本里解析命令行参数：

import sys
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
df = spark.read.table("your_source_table")

# 提取目标环境参数
target_env = "dev"  # 设置默认值
for idx in range(len(sys.argv)):
    if sys.argv[idx] == "--env" and idx + 1 < len(sys.argv):
        target_env = sys.argv[idx+1]

# 生成对应环境的schema路径
schema_string = f"your_catalog.{target_env}_schema"

# 写入数据到目标schema
df.write.mode("overwrite").saveAsTable(f"{schema_string}.your_target_table")

方案2：通过作业环境变量传递 #

1. 在bundle.yml的任务中配置环境变量：

jobs:
  your-job-name:
    tasks:
      - task_key: write-to-schema
        spark_python_task:
          python_file: "/path/to/your/script.py"
        env_vars:
          TARGET_ENV: "{{ bundle.target_env }}"

2. 在Python脚本中读取环境变量：

import os
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
df = spark.read.table("your_source_table")

# 获取环境变量，默认用dev兜底
target_env = os.getenv("TARGET_ENV", "dev")

# 确定目标schema
schema_string = f"your_catalog.{target_env}_schema"

# 执行写入操作
df.write.mode("overwrite").saveAsTable(f"{schema_string}.your_target_table")

验证方法 #

运行指定环境的bundle命令即可：

databricks bundle run -t dev your-job-name

脚本会自动识别dev作为目标环境，写入对应的schema。

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