精通 Databricks API：大规模处理数据管道

Databricks API 使您能够以编程方式与 Databricks 交互，从而自动化工作流、管理集群、执行作业并访问数据。这些 API 支持通过个人访问令牌、OAuth 或 Azure Active Directory 进行身份验证。

在本文中，我将带您深入、动手实践地了解 Databricks REST API。无论您是 Databricks 新手，还是想优化现有工作流，本指南都将帮助您掌握关键操作，包括身份验证、作业管理以及与外部系统集成。 

在阅读完本文后，您将清楚掌握以下蓝图：

• 使用个人访问令牌设置安全的 API 身份验证。  
• 使用 API 端点创建、更新并运行作业。 
• 应用最佳实践以提升性能、优化成本并处理错误。 
• 将 Databricks 与其他平台和编排工具集成。

如果您完全是 Databricks 新手并希望快速上手，请阅读我们的博文 How to Learn Databricks: A Beginner’s Guide to the Unified Data Platform，它将帮助您理解 Databricks 的核心功能与应用，并为您的学习提供结构化路径。

为什么要使用 Databricks API？

Databricks REST API 为开发者和数据工程师解锁强大能力，使其能够以编程方式管理资源并简化工作流。以下是 Databricks API 发挥巨大作用的方式：

降低作业调度与集群管理的人工开销

Databricks API 允许用户自动化作业调度和集群管理等任务。这降低了管理这些流程所需的人工投入，释放时间用于更具战略性的活动。通过自动化重复性任务，组织可以提升效率并减少人为错误的可能性，避免代价高昂的失误或停机。

与外部系统的无缝集成

Databricks API 具备灵活性，能够与外部平台集成，例如编排工具 Apache Airflow 或 Azure Data Factory。

您还可以将 Databricks API 用于 Prometheus 或 Datadog 等监控与告警系统。Databricks API 也常用于来自 Kafka 或 RESTful 数据服务的摄取管道。这些集成有助于构建端到端工作流，跨多个系统无缝衔接数据工程与分析。

支持成千上万条自动化管道的可扩展性

Databricks API 支持创建和管理成千上万条自动化管道。它可轻松扩展，以并行运行海量作业，并根据工作负载需求动态扩缩集群。此类可扩展性确保您的基础设施能够在企业级规模下无缝适配业务需求。

核心 API 操作与示例

Databricks REST API 提供丰富操作来管理作业、集群、文件和访问权限。下面，我将通过实用示例讲解关键 API 操作，帮助您构建并自动化稳健的工作流。

作业（Jobs）

作业是 Databricks 中自动化工作流的基石。API 可让您无缝创建、管理与监控作业。

创建作业

使用 POST /API/2.1/jobs/create 端点定义新作业。如下 JSON 载荷将创建名为 My Job 的 Databricks 新作业，并运行位于 /path/to/notebook 的笔记本。该载荷使用已有 ID 为 cluster-id 的集群，避免新建。如果作业失败，将向 [email protected] 发送邮件通知。

{
  "name": "My Job",
  "existing_cluster_id": "cluster-id",
  "notebook_task": {
    "notebook_path": "/path/to/notebook"
  },
  "email_notifications": {
    "on_failure": ["[email protected]"]
  }
}技术支持

列出并获取作业详情

您可以使用 GET /api/2.1/jobs/list 端点检索工作区中所有作业列表。如下示例命令将列出全部作业：

curl -n -X GET \
https://<databricks-instance>.cloud.databricks.com/api/2.1/jobs/list技术支持

触发运行

使用 POST /api/2.1/jobs/run-now 端点立即触发某个作业的运行。

curl -n -X POST \
https://<databricks-instance>.cloud.databricks.com/api/2.1/jobs/run-now?job_id=123456789技术支持

删除作业与运行

尽管最新 API 版本没有直接删除作业的端点，但您可以通过其他端点管理作业运行及其生命周期。使用删除作业：POST /api/2.1/jobs/delete 端点删除作业，使用 POST /api/2.1/jobs/runs/cancel 取消某次运行。

监控运行、错误处理与日志

使用 GET /api/2.1/jobs/runs/get 监控作业状态。通过 run_state 字段检索日志或处理错误。

集群（Clusters）

集群为作业运行提供计算资源。API 简化了集群全生命周期管理。

创建与编辑集群

使用 POST /api/2.1/clusters/create 创建集群。例如，您可以在请求体中包含以下内容。

{
  "cluster_name": "Example Cluster",
  "spark_version": "12.2.x-scala2.12",
  "node_type_id": "i3.xlarge",
  "num_workers": 4
}技术支持

然后，运行以下示例命令以创建该集群。

curl -n -X POST \
-H 'Content-Type: application/json' \
-d '{"cluster_name": "My Cluster", "spark_version": "7.3.x-scala2.12", "node_type_id": "i3.xlarge", "num_workers": 2}' \
https://<databricks-instance>.cloud.databricks.com/api/2.1/clusters/create技术支持

启动、重启、调整规模与终止集群

使用以下 API：

• 启动集群：POST /api/2.1/clusters/start

• 重启集群： POST /api/2.1/clusters/restart

• 调整规模： PATCH /api/2.1/clusters/edit

• 终止集群： POST /api/2.1/clusters/delete

短期集群 vs. 通用集群的最佳实践

请参考以下最佳实践：

• 短期（Ephemeral）集群： 适用于需要动态资源分配并可在作业完成后终止的作业。
• 通用（All-Purpose）集群： 适合持续性工作负载，在多个作业之间复用集群。

DBFS（Databricks 文件系统）

DBFS 支持在 Databricks 中进行文件存储与检索。

上传与下载文件

您可以使用 POST /api/2.1/dbfs/upload 以分块（多部分）方式上传文件至 DBFS，或在处理大文件（>1MB）时使用。使用 GET /api/2.1/dbfs/read 下载文件。

列出目录并处理大型文件操作

使用 GET /api/2.1/dbfs/list 探索目录并以编程方式管理文件。对于大文件上传，使用 POST /api/2.1/dbfs/put 的分块上传并配合 overwrite 参数。

机密与身份验证

Secrets API 确保敏感信息的安全存储与检索。

使用 Secrets API 安全存储凭据

使用 POST /api/2.1/secrets/create 端点创建用于存储敏感信息的新机密。

例如，您可以按如下方式创建 JSON 请求体：

{
  "scope": "my-scope",
  "key": "my-key",
  "string_value": "my-secret-value"
}技术支持

然后，使用以下命令创建机密：

curl -n -X POST \
-H 'Content-Type: application/json' \
-d '{"scope": "my-scope", "key": "my-key", "string_value": "my-secret-value"}' \
https://<databricks-instance>.cloud.databricks.com/api/2.1/secrets/create技术支持

为集成检索机密

使用 GET /API/2.1/secrets/get 端点检索用于外部集成的机密。

权限与访问控制

Databricks 平台可对作业与资源进行精细化访问控制。

设置与更新权限

使用 POST /api/2.1/permissions/jobs/update 端点分配所有者或查看者等角色。如下示例载荷用于更新权限：

{
  "access_control_list": [
    {
      "user_name": "[email protected]",
      "permission_level": "CAN_MANAGE"
    }
  ]
}技术支持

基于角色的访问控制（RBAC）

使用 Databricks 内置的基于角色的访问控制（RBAC），在组织内不同角色之间管理权限。

Databricks API 速率限制、错误处理与故障排查

使用 Databricks REST API 时，务必考虑速率限制、优雅地处理错误，并解决常见问题，以确保性能可靠、运行顺畅。本节将介绍优化 API 使用与高效排障的策略。

速率限制

Databricks API 实施速率限制以确保公平使用并维持服务可靠性。 

• 请求频率的官方指南：此限制会约束用户每分钟可发起的 API 调用次数。超出将导致 HTTP 429 错误。每天请求数（RPD）会限制用户在 24 小时内可发起的总请求次数。
• 处理 429 错误与退避策略：实施带指数退避的重试机制，即在重试之间逐步延长延迟。这有助于避免压垮服务器，并降低再次触发速率限制的可能性。同时，监控请求模式，尽可能合并高频调用，以减少总体请求量。

常见问题与解决方案

以下是使用 Databricks API 时常见的问题及其解决方法。

• 身份验证陷阱：此错误常由令牌无效或过期引起。解决方法是确保令牌有效并正确配置。验证令牌有效性，必要时重新生成。确保服务主体在 Azure Databricks 中配置正确并授予适当角色。
• 数据格式或 JSON 错误：在发送 API 请求前务必校验 JSON 数据。使用 try-catch 块优雅地处理解析错误。利用 JSON 校验工具在发送请求前检查数据格式也有助于解决错误。
• 调试作业失败：使用 Databricks API 检索作业日志以进行调试。实施健全的日志与监控，快速识别并处理作业失败。

性能瓶颈

若遇到性能瓶颈，建议考虑以下方案：

• 最小化重复调用的开销：对高频访问数据实施缓存以减少 API 调用次数。同时使用批处理操作，将多个请求合并为一次调用以降低开销。
• 使用缓存或批处理操作：将高频访问数据存入缓存层，减少重复 API 调用。也可以将多个请求合并为单个 API 调用以降低开销并提升性能。

高级工作流与并行处理

Databricks REST API 十分灵活，既支持简单的单线程工作流，也支持复杂的并行处理。通过利用高级工作流，您可以优化效率并与外部系统无缝集成。下面来看如何实现这些功能。

请注意，务必按如下方式配置环境变量：

• DATABRICKS_HOST：该变量存储您的 Databricks 实例 URL，用于构建 API 请求。

• DATABRICKS_TOKEN：该变量保存您的访问令牌，使在 API 调用中包含令牌更为便捷。

用例 1：单线程 API 摄取

顺序调用单个 API 端点适用于小型任务或简单自动化。以下是使用 Python requests 库的实用示例。

import requests
import os

# Set up environment variables
DATABRICKS_HOST = os.getenv("DATABRICKS_HOST")
DATABRICKS_TOKEN = os.getenv("DATABRICKS_TOKEN")

# API request to list jobs
url = f"{DATABRICKS_HOST}/api/2.1/jobs/list"
headers = {"Authorization": f"Bearer {DATABRICKS_TOKEN}"}

response = requests.get(url, headers=headers)
if response.status_code == 200:
    jobs = response.json().get("jobs", [])
    for job in jobs:
        print(f"Job ID: {job['job_id']}, Name: {job['settings']['name']}")
else:
    print(f"Error: {response.status_code} - {response.text}")技术支持

用例 2：并行或分布式 API 调用

对于更大规模的数据摄取或需同时处理多个端点的场景，并行处理能显著提升效率。以下示例展示如何使用 Apache Spark 并发分发 API 调用。

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
import requests

# Initialize Spark session
spark = SparkSession.builder.appName("ParallelAPI").getOrCreate()

# Sample data for parallel calls
data = [{"job_id": 123}, {"job_id": 456}, {"job_id": 789}]
df = spark.createDataFrame(data)

# Define API call function
def fetch_job_details(job_id):
    url = f"{DATABRICKS_HOST}/api/2.1/jobs/get"
    headers = {"Authorization": f"Bearer {DATABRICKS_TOKEN}"}
    response = requests.get(url, headers=headers, params={"job_id": job_id})
    if response.status_code == 200:
        return response.json().get("settings", {}).get("name", "Unknown")
    return f"Error: {response.status_code}"

# Register UDF
fetch_job_details_udf = udf(fetch_job_details)

# Apply UDF to DataFrame
results_df = df.withColumn("job_name", fetch_job_details_udf("job_id"))
results_df.show()技术支持

请考虑以下有关可扩展性与成本权衡的建议。

• 可扩展性：Spark 允许通过增加集群节点实现横向扩展，尤其适合处理大批量 API 请求。
• 成本权衡：并行处理可缩短执行时间，但会增加资源使用与成本。请确保集群配置在性能与成本之间取得平衡。
• 集群配置建议：根据工作负载在 CPU 与内存之间平衡选择节点。确保有足够的工作节点以处理请求量，同时避免压垮集群。优化 Spark 设置（如 spark.executor.memory 与 spark.driver.memory）以防止内存问题。

编排与集成

将 Databricks 与 Airflow 或 Dagster 等外部编排工具集成，可以管理复杂工作流并在不同系统间自动化作业执行。以下示例展示如何使用 Airflow 触发 Databricks 作业。

from airflow import DAG
from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator
from datetime import datetime

# Define default arguments for the DAG
default_args = {
    "owner": "airflow",  # Owner of the DAG
    "depends_on_past": False,  # Do not depend on past DAG runs
    "retries": 1,  # Number of retry attempts in case of failure
}

# Define the DAG
with DAG(
    dag_id="databricks_job_trigger",  # Unique DAG ID
    default_args=default_args,  # Apply default arguments
    schedule_interval=None,  # Manually triggered DAG (no schedule)
    start_date=datetime(2023, 1, 1),  # DAG start date
) as dag:

    # Task to trigger a Databricks job
    run_job = DatabricksRunNowOperator(
        task_id="run_databricks_job",  # Unique task ID
        databricks_conn_id="databricks_default",  # Connection ID for Databricks
        job_id=12345,  # Replace with the actual Databricks job ID
    )

    # Set task dependencies (if needed)
    run_job技术支持

对于 CI/CD 流水线，将 Git 与 Databricks 集成等工具可以自动化在不同环境间部署代码与资产。这样可确保变更在投入生产前得到测试与验证。

# Example of using Databricks REST API in a CI/CD pipeline
import requests

def deploy_to_production():
    # Assuming you have a personal access token
    token = "your_token_here"
    headers = {"Authorization": f"Bearer {token}"}
    
    # Update job or cluster configurations as needed
    url = "https://your-databricks-instance.cloud.databricks.com/api/2.1/jobs/update"
    payload = {"job_id": "123456789", "new_settings": {"key": "value"}}
    
    response = requests.post(url, headers=headers, json=payload)
    if response.status_code == 200:
        print("Deployment successful.")
    else:
        print(f"Deployment failed. Status code: {response.status_code}")技术支持

实践示例：创建多步骤管道

了解了 Databricks API 的不同应用后，下面通过一个自动化管道示例带您实操。该管道将从外部 API 摄取数据，在 Databricks 中转换数据，然后将结果写入 Databricks 表。这是现代数据工程任务的典型端到端工作流。若您需要回顾 Databricks 的主要功能（包括数据摄取），建议阅读我们的教程 7 Must-know Concepts For Any Data Specialist。

通过个人访问令牌进行身份验证

要与 Databricks REST API 交互，您需要使用个人访问令牌（PAT）进行身份验证。按以下步骤生成 PAT。 

• 登录您的 Databricks 工作区。

Databricks 工作区。作者供图。

• 前往 Settings > User Settings > Access Tokens。

在 Databricks 中生成访问令牌。作者供图。

• 生成新令牌并妥善保存。

如果您需要回顾 Databricks SQL，建议阅读我们的 Databricks SQL 教程，了解如何通过 Databricks Web 界面设置 SQL Warehouse。

使用以下任一方式设置环境变量：

使用终端

export DATABRICKS_HOST=https://<your-databricks-instance>
export DATABRICKS_TOKEN=<your-access-token>技术支持

使用 Python 的 requests 库进行 API 调用

import os
import requests

DATABRICKS_HOST = os.getenv("DATABRICKS_HOST")
DATABRICKS_TOKEN = os.getenv("DATABRICKS_TOKEN")

headers = {"Authorization": f"Bearer {DATABRICKS_TOKEN}"}技术支持

设置包含多个任务的 Databricks 作业

一个 Databricks 作业可包含多个任务，每个任务完成管道中的一个步骤。本例包括：

• 任务 1：从外部 API 获取数据并写入 DBFS。
• 任务 2：在 Databricks 笔记本中转换数据。
• 任务 3：将转换后的数据写入 Delta 表。

以下示例使用 POST /api/2.1/jobs/create 端点创建作业。

{
  "name": "Multi-Step Pipeline",
  "tasks": [
    {
      "task_key": "fetch_data",
      "existing_cluster_id": "your-cluster-id",
      "notebook_task": {
        "notebook_path": "/Users/example_user/fetch_data"
      }
    },
    {
      "task_key": "transform_data",
      "depends_on": [{"task_key": "fetch_data"}],
      "existing_cluster_id": "your-cluster-id",
      "notebook_task": {
        "notebook_path": "/Users/example_user/transform_data"
      }
    },
    {
      "task_key": "write_results",
      "depends_on": [{"task_key": "transform_data"}],
      "existing_cluster_id": "your-cluster-id",
      "notebook_task": {
        "notebook_path": "/Users/example_user/write_results"
      }
    }
  ]
}技术支持

触发工作流并检索作业结果/输出

使用 POST /api/2.1/jobs/run-now 触发作业

curl -n -X POST \
https://$DATABRICKS_HOST/api/2.1/jobs/run-now?job_id=your_job_id技术支持

使用 GET /api/2.1/jobs/runs/get 携带 run_id 监控作业状态并跟踪进度。

response = requests.get(f"{DATABRICKS_HOST}/api/2.1/jobs/runs/get", headers=headers, params={"run_id": run_id})
print(response.json())技术支持

通过 GET /api/2.1/jobs/runs/get-output 访问任务的输出日志或结果。

日志与错误处理

在日志方面，可将作业日志写入 DBFS 或 Datadog、Splunk 等监控工具。同时，在每个任务的笔记本中加入详细日志，提升可追溯性。

同样地，在笔记本中实现 try-except 块来捕获并记录异常。使用 Databricks Jobs API 检索日志以调试失败的运行。以下示例展示笔记本中的错误处理。

try:
    # Code to fetch data from API
    response = requests.get(api_url)
    # Process data
except Exception as e:
    print(f"An error occurred: {e}")
    # Log error and exit技术支持

您也可以通过 Databricks Alerts API 或第三方工具集成邮件或 Slack 通知。

最佳实践与建议

在 Databricks 中实施最佳实践，可显著提升安全性、降低成本并改善整体系统可靠性。以下是使用 Databricks API 时应重点考虑的关键策略与领域。

安全与治理

避免在代码或笔记本中明文存储令牌。请使用 Databricks Secret 通过 Databricks Secrets API 安全存储 API 密钥或凭据，确保敏感数据不以明文形式出现在代码或笔记本中。始终使用环境变量将令牌或机密传递给脚本，但要确保安全管理这些变量，并避免将其提交到版本控制。

使用基于角色的访问控制（RBAC）在工作区、集群与作业层面定义权限，并将对敏感 API 的访问限制在必要的用户或服务账户范围内。

成本管理

为有效控制成本，可考虑将单节点集群用于持续、低负载任务；在未充分利用时，它们可能更具成本效益。此外，短期作业集群会为作业自动创建与终止，避免资源空置。

由于云资源采用动态定价，使用非高峰时段调度可显著降低成本。使用 Jobs API 在更具成本效益的时间窗口定义作业计划。

最后，通过定期检查集群利用率来监控集群使用情况，确保资源不被浪费。配置集群在空闲达到指定时长后自动终止，以避免不必要的开销。

持续测试与监控

对于自动化测试与作业定义，使用单元测试在作业逻辑上实施测试，及早发现错误。监控作业运行并记录性能指标，利用 Jobs API 以编程方式跟踪作业运行并分析历史运行数据：

结论与后续步骤

使用 Databricks REST API 能帮助团队自动化工作流、优化成本并与外部系统无缝集成，重塑数据管道与分析的管理方式。建议您先从测试一个基础 API 调用（如创建作业）开始，以逐步建立对平台的信心。您也可以进一步探索如 Unity Catalog 的精细化治理等高级主题，或深入实时流处理用例。

此外，建议参考 Official Databricks REST API Docs 解答尚存疑问，并探索 GitHub 上的 Databricks CLI 与 Databricks SDK for Python 仓库，了解更流畅的开发方式。

如果您想系统学习 Databricks 的基础概念，强烈推荐参加我们的 Introduction to Databricks 课程。该课程将带您了解将 Databricks 作为商业智能的数据仓库解决方案。我也建议阅读我们的 Databricks Certifications In 2024 博文，了解如何获取 Databricks 认证、探索职业收益，并为您的职业目标选择合适的认证。