如何从外部应用启动 Databricks cluster 并传递配置参数？

可以通过 Databricks Jobs API 实现。外部应用后端接收用户选择的配置档位，将其映射到相应的计算资源配置，再提交数据处理任务。批处理通常可以交给 Jobs 管理集群和任务的生命周期：运行时创建 job compute，任务结束后释放资源。

常见做法有两种：

• 配置档位固定时，提前创建计算配置不同的 Job。外部应用选择对应的 job_id，调用 jobs/run-now。
• 每次运行都需要生成计算配置时，调用 jobs/runs/submit，在请求中同时提供 new_cluster 和任务定义。

50 GB、100 GB、1000 GB 可以作为分档依据，但仅凭数据量无法确定准确的节点数量。最终配置仍需用实际工作负载验证。

一、架构如何设计

应用可以按以下流程调用：

用户选择数据和配置档位
    ↓
外部应用后端：校验数据权限、档位和请求参数
    ↓
将档位映射到 job_id 或 new_cluster 配置
    ↓
调用 Databricks Jobs API
    ↓
Databricks 创建计算资源并执行任务
    ↓
应用保存 run_id，查询状态并展示结果

数据应提前存放在 Databricks 能访问的位置，例如云对象存储、Unity Catalog Volume 或表中。API 请求只传数据位置和处理参数，不传几十 GB 的数据内容。

档位数量有限时，可以在后端维护一张配置表：

这些数据量范围只是业务分档示例。节点类型、worker 数量和扩缩容范围应根据压测结果确定。

前端只需提交档位标识，具体的 job_id、节点类型和资源上限由后端管理，避免用户绕过界面提交任意配置。

二、需要哪些 API 和服务

应用需要 Databricks Workspace 和 Jobs API，可以直接发送 HTTPS 请求，也可以使用 Databricks SDK，无需额外引入调度服务。

应用通过接口返回的 run_id 跟踪后续运行情况。Jobs API：触发运行、运行管理

如果需要单独管理已有集群，可以使用 Clusters API。clusters/start 用于启动已经终止的集群，并保留原来的配置。创建或启动集群本身不会执行数据处理代码，按 Job 创建的集群也不能通过该接口重新启动。Clusters API

三、外部应用需要传递哪些参数

不同用途的参数放在不同位置：

业务参数和计算配置容易混淆。把 "size": "large" 放进 job_parameters，只会将这个值传给任务代码，不会自动改变集群规模。jobs/run-now 没有用于任意覆盖 new_cluster 的请求字段。因此，固定档位可以映射到不同 Job；每次生成计算配置时，则使用 runs/submit。Run now 请求字段

四、固定档位：预建 Job 后触发运行

对于固定的配置菜单，可以创建多个使用相同 Notebook 或 Python wheel 的 Job，为每个 Job 配置不同的 classic job compute。

部署时还应配置运行身份、依赖、超时和并发上限。不要在每次收到用户请求后，先修改同一个 Job 的计算配置再运行，否则并发请求会争用这份共享配置。

后端调用示例

使用以下 Python requests 示例前，需要满足这些条件：

• 已创建各个档位对应的 Job，并配置好计算资源。
• Job 中的 Notebook 接收 input_path、output_table 和 request_id。
• 服务主体已获得触发 Job 的权限，Job 的运行身份也具有所需的数据访问权限。
• 环境变量由后端部署配置或密钥管理服务提供。

自动化调用可以使用服务主体的 OAuth M2M 认证。Workspace 级 token endpoint 为 /oidc/v1/token。如果使用 SDK，也可以通过统一认证管理 token 的获取和刷新。OAuth M2M 官方文档

import os
import requests

host = os.environ["DATABRICKS_HOST"].rstrip("/")

jobs_by_profile = {
    "small": int(os.environ["DB_JOB_SMALL"]),
    "medium": int(os.environ["DB_JOB_MEDIUM"]),
    "large": int(os.environ["DB_JOB_LARGE"]),
    "xlarge": int(os.environ["DB_JOB_XLARGE"]),
}

# 示例业务请求；实际应用中需先完成权限和参数校验。
profile = "medium"
request_id = "batch-20261104-001"

token_response = requests.post(
    f"{host}/oidc/v1/token",
    auth=(
        os.environ["DATABRICKS_CLIENT_ID"],
        os.environ["DATABRICKS_CLIENT_SECRET"],
    ),
    data={
        "grant_type": "client_credentials",
        "scope": "all-apis",
    },
    timeout=30,
)
token_response.raise_for_status()
access_token = token_response.json()["access_token"]

run_response = requests.post(
    f"{host}/api/2.2/jobs/run-now",
    headers={"Authorization": f"Bearer {access_token}"},
    json={
        "job_id": jobs_by_profile[profile],
        "job_parameters": {
            "input_path": "/Volumes/main/landing/raw/batch_001/",
            "output_table": "main.processed.batch_001",
            "request_id": request_id,
        },
        "idempotency_token": request_id,
    },
    timeout=30,
)
run_response.raise_for_status()

run_id = run_response.json()["run_id"]
print({"request_id": request_id, "run_id": run_id})

idempotency_token 最多为 64 个字符。同一次逻辑提交因网络错误而重试时，应复用原来的 token；需要重新执行任务时，则使用新的 token。应用还应保存业务请求与 run_id 的对应关系。幂等提交说明

Notebook 如何读取参数

Notebook 可以通过 dbutils.widgets.get 读取参数。Job 级参数会自动传递给接受键值参数的任务；采用参数数组的任务类型则需要显式引用参数，不能直接套用 Notebook 的方式。Job 参数传递规则

以下示例读取 Parquet 数据，并将其写成 Unity Catalog 中的 Delta 表。读取和写入之间的代码可以替换为实际处理逻辑：

dbutils.widgets.text("input_path", "")
dbutils.widgets.text("output_table", "")
dbutils.widgets.text("request_id", "")

input_path = dbutils.widgets.get("input_path")
output_table = dbutils.widgets.get("output_table")

if not input_path or not output_table:
    raise ValueError("input_path and output_table are required")

df = spark.read.parquet(input_path)

df.write.format("delta").mode("errorifexists").saveAsTable(output_table)

示例使用 errorifexists 避免覆盖已有表。生产任务需要根据业务设计重复执行策略，例如按批次写入、分区覆盖或使用 MERGE。

五、逐次生成配置：提交 new_cluster

如果同一任务每次都需要不同的节点类型或 worker 范围，可以向 POST /api/2.2/jobs/runs/submit 提交以下请求：

{
  "run_name": "external-data-processing",
  "idempotency_token": "batch-20261104-002",
  "timeout_seconds": 7200,
  "tasks": [
    {
      "task_key": "process_data",
      "new_cluster": {
        "spark_version": "<SUPPORTED_RUNTIME_ID>",
        "node_type_id": "<AVAILABLE_NODE_TYPE_ID>",
        "policy_id": "<COMPUTE_POLICY_ID>",
        "apply_policy_default_values": true,
        "autoscale": {
          "min_workers": 2,
          "max_workers": 8
        }
      },
      "notebook_task": {
        "notebook_path": "/Shared/process_data",
        "base_parameters": {
          "input_path": "/Volumes/main/landing/raw/batch_002/",
          "output_table": "main.processed.batch_002",
          "request_id": "batch-20261104-002"
        }
      }
    }
  ]
}

示例中的 2 至 8 个 worker 和两小时超时仅用于说明请求结构，不是针对某个数据量的推荐配置。实际使用前，需要替换占位符，并根据云平台、计算策略和数据访问要求补齐配置。

如果 worker 数量固定，使用 num_workers，不要同时设置 autoscale。spark_version 和 node_type_id 应选择目标 Workspace 支持的值。也可以先在界面中配置计算资源，再复制 JSON，减少字段错误。计算配置 API

runs/submit 不会保存可重复触发的 Job。如果工作负载需要长期维护、配置重试和统一管理，通常更适合预建 Job。一次性运行接口

六、如何调整规模和控制成本

用实测结果确定档位

同样是 100 GB 数据，简单过滤与大表 JOIN 的资源需求可能相差很大。配置时还要考虑文件数量、压缩格式、数据倾斜、shuffle、CPU 使用率，以及内存不足时写入磁盘的 spill。

建议用有代表性的数据测试几个档位，记录执行时间、资源利用率和整次运行成本。内存需求主要取决于同时处理的数据和中间结果，集群内存不必容纳全部输入数据。计算规模选择建议

区分档位选择与运行时扩缩容

应用选择的档位决定本次运行的节点类型和资源边界，autoscale 在这个边界内调整 worker 数量。

Autoscaling 不会因为应用传入 data_size_gb=200 就直接切换到某个固定规模，也不能保证解决数据倾斜或 driver 内存不足的问题。对于短任务，还需要比较固定规模和自动扩缩容的实际耗时。Autoscaling 配置

结合启动时间评估成本

• 批处理优先使用 job compute。单任务计算资源在任务结束后终止，共享 job compute 在最后一个使用它的任务结束后终止。Job compute 生命周期
• 如果任务受支持且 Workspace 可用，可以评估 serverless jobs compute。Databricks 会管理其计算配置，适合希望减少节点配置维护的场景。任务计算类型
• 对频繁运行且启动延迟敏感的 classic 工作负载，可以评估 instance pools。预留的空闲实例也占用云资源，因此需要同时控制最小空闲数和释放设置。Pool 配置

比较成本时，应计算整次任务的开销。较小的集群如果运行时间明显更长，总成本未必更低。

七、上线前需要处理什么

凭据应保存在应用后端。调用 API 的身份与 Job 实际执行时使用的身份可能不同，两者都需要相应权限。传入数据路径不会自动授予读取权限。

后端还应校验输入路径、输出表和配置档位，确保用户只能处理自己获准访问的数据。计算策略可以约束节点类型、worker 上限和允许的配置。

拿到 run_id 只表示提交成功。应用还需要持续查询运行结果，处理排队、启动失败、执行失败和取消等情况，并为查询请求设置合理的退避间隔。查询运行接口

API 提交幂等与数据写入幂等需要分别处理。idempotency_token 可以避免重复提交产生多个运行，但任务重试或人为重跑仍可能重复写入数据，输出逻辑需要单独处理这种情况。

备注：内容仅供参考。