如何自动化执行Databricks Notebook中的Python脚本？

问题背景 #

我目前使用以下Python脚本从CSV文件填充Databricks中的数据表，目标是将CSV数据导入已有表格并匹配各列数据类型。现在希望自动化该流程，无需手动打开Notebook运行脚本，想了解：

• 是否能用SQL实现后通过编程调用？
• 有没有Databricks REST API可执行Python脚本？
• 还有其他自动化方式吗？

原Python脚本：

tables = [
'care_site',
]

target_db = 'demo_cdm'
data_path = '/FileStore/tables/prod/ohdsi/demo_cdm/'

for t in tables:
  tgt_t = t.lower()
  df = spark.sql('SELECT * FROM {db}.{table}'.format(db=target_db, table=tgt_t))
  spark.read.options(delimiter=",", header="True", dateFormat="yyyy-MM-dd")\
          .schema(df.schema)\
          .csv(data_path + t + '/' + t + '.csv')\
          .write.format('delta')\
          .insertInto(target_db + '.' + tgt_t.lower(), overwrite=True)
  spark.sql('REFRESH TABLE {db}.{table}'.format(db=target_db, table=tgt_t))

一、用SQL替代Python脚本，再编程调用 #

可以用COPY INTO SQL命令实现需求（解决之前的类型转换问题），再通过工具自动化执行：

1. 编写SQL脚本 #

单表示例（自动复用目标表Schema，匹配列类型）：

COPY INTO demo_cdm.care_site
FROM '/FileStore/tables/prod/ohdsi/demo_cdm/care_site/care_site.csv'
FILEFORMAT = CSV
OPTIONS (
  header = 'true',
  delimiter = ',',
  dateFormat = 'yyyy-MM-dd'
)
COPY_OPTIONS (
  overwrite = 'true'
);

REFRESH TABLE demo_cdm.care_site;

批量处理多表可通过Databricks SQL变量或存储过程实现循环逻辑。

2. 编程调用SQL的方式 #

• Databricks CLI：执行本地SQL文件

  databricks sql execute --file ./import_tables.sql --warehouse-id <你的仓库ID>
  

• REST API：调用/api/2.0/sql/statements端点提交SQL任务，可获取执行状态和结果。

二、用Databricks REST API执行Python脚本 #

有两种方式通过API执行Python逻辑：

1. 提交Notebook任务 #

调用/api/2.0/jobs/runs/submit端点，指定包含Python代码的Notebook路径，示例请求体：

{
  "notebook_task": {
    "notebook_path": "/Workspace/import_csv_to_tables"
  },
  "new_cluster": {
    "spark_version": "13.3.x-scala2.12",
    "node_type_id": "Standard_DS3_v2",
    "num_workers": 1
  }
}

若未预先创建作业，可直接提交一次性运行任务。

2. 直接执行Python代码 #

通过spark_python_task直接传入代码，无需依赖Notebook：

{
  "new_cluster": {
    "spark_version": "13.3.x-scala2.12",
    "node_type_id": "Standard_DS3_v2",
    "num_workers": 1
  },
  "spark_python_task": {
    "python_code": "tables = ['care_site']\ntarget_db = 'demo_cdm'\ndata_path = '/FileStore/tables/prod/ohdsi/demo_cdm/'\nfor t in tables:\n  tgt_t = t.lower()\n  df = spark.sql('SELECT * FROM {db}.{table}'.format(db=target_db, table=tgt_t))\n  spark.read.options(delimiter=\",\", header=\"True\", dateFormat=\"yyyy-MM-dd\").schema(df.schema).csv(data_path + t + '/' + t + '.csv').write.format('delta').insertInto(target_db + '.' + tgt_t.lower(), overwrite=True)\n  spark.sql('REFRESH TABLE {db}.{table}'.format(db=target_db, table=tgt_t))"
  }
}

三、其他自动化方式 #

• Databricks作业调度：在Databricks UI创建作业，选择Notebook或Python任务，设置定时触发（如每日/每周），无需手动干预。
• 外部编排工具：通过Airflow、Azure Data Factory、AWS Step Functions等工具调用Databricks API/CLI，适合复杂数据流编排场景。
• Databricks Workflows：使用原生工作流功能构建多任务Pipeline，支持依赖管理和调度，实现端到端自动化。

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