databricks-python-sdk

Databricks 開発ガイド

このスキルでは、Databricks SDK、Databricks Connect、CLI、REST API の使用方法を説明します。

SDK ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/ GitHub リポジトリ： https://github.com/databricks/databricks-sdk-py

---

環境のセットアップ

• 既存の仮想環境 .venv を使用するか、uv で作成します
• Spark 操作の場合：uv pip install databricks-connect
• SDK 操作の場合：uv pip install databricks-sdk
• Databricks CLI のバージョンは 0.278.0 以上にしてください

設定

• デフォルトのプロファイル名：DEFAULT
• 設定ファイル：~/.databrickscfg
• 環境変数：DATABRICKS_HOST、DATABRICKS_TOKEN

---

Databricks Connect（Spark 操作）

Databricks クラスターに対してローカルで Spark コードを実行するには、databricks-connect を使用します。

from databricks.connect import DatabricksSession

# Auto-detects 'DEFAULT' profile from ~/.databrickscfg
spark = DatabricksSession.builder.getOrCreate()

# With explicit profile
spark = DatabricksSession.builder.profile("MY_PROFILE").getOrCreate()

# Use spark as normal
df = spark.sql("SELECT * FROM catalog.schema.table")
df.show()

重要： .master("local[*]") は設定しないでください。Databricks Connect で問題が発生します。

---

REST API への直接アクセス

SDK にまだ実装されていない操作や、SDK 経由では複雑すぎる操作には、REST API に直接アクセスします。

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Direct API call using authenticated client
response = w.api_client.do(
    method="GET",
    path="/api/2.0/clusters/list"
)

# POST with body
response = w.api_client.do(
    method="POST",
    path="/api/2.0/jobs/run-now",
    body={"job_id": 123}
)

使用する場合： 利用可能な場合は SDK メソッドを優先してください。次の場合は api_client.do を使用します。

• SDK にまだ実装されていない新しい API エンドポイント
• SDK の抽象化では問題がある複雑な操作
• 生の API レスポンスのデバッグ／テスト

---

Databricks CLI

# Check version (should be >= 0.278.0)
databricks --version

# Use specific profile
databricks --profile MY_PROFILE clusters list

# Common commands
databricks clusters list
databricks jobs list
databricks workspace list /Users/me

---

SDK リファレンス

API ごとのメソッドシグネチャとドキュメントへの直接リンクをまとめた表（Clusters、Jobs、SQL warehouses、Unity Catalog、Serving、Vector Search など）については、references/doc-index.md を参照してください。主な API 領域の実例は examples/ にあります。

SDK ドキュメントの構成

SDK ドキュメントの URL には、予測可能なパターンがあります。

Base: https://databricks-sdk-py.readthedocs.io/en/latest/

Workspace APIs:  /workspace/{category}/{service}.html
Account APIs:    /account/{category}/{service}.html
Authentication:  /authentication.html
DBUtils:         /dbutils.html

Workspace API カテゴリ

カテゴリ	サービス
compute	clusters, cluster_policies, command_execution, instance_pools, libraries
catalog	catalogs, schemas, tables, volumes, functions, storage_credentials, external_locations
jobs	jobs
sql	warehouses, statement_execution, queries, alerts, dashboards
serving	serving_endpoints
vectorsearch	vector_search_indexes, vector_search_endpoints
pipelines	pipelines
workspace	repos, secrets, workspace, git_credentials
files	files, dbfs
ml	experiments, model_registry

---

認証

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/authentication.html

環境変数

DATABRICKS_HOST=https://your-workspace.cloud.databricks.com
DATABRICKS_TOKEN=dapi...  # Personal Access Token

コードパターン

# Auto-detect credentials from environment
from databricks.sdk import WorkspaceClient
w = WorkspaceClient()

# Explicit token auth
w = WorkspaceClient(
    host="https://your-workspace.cloud.databricks.com",
    token="dapi..."
)

# Azure Service Principal
w = WorkspaceClient(
    host="https://adb-xxx.azuredatabricks.net",
    azure_workspace_resource_id="/subscriptions/.../resourceGroups/.../providers/Microsoft.Databricks/workspaces/...",
    azure_tenant_id="tenant-id",
    azure_client_id="client-id",
    azure_client_secret="secret"
)

# Use a named profile from ~/.databrickscfg
w = WorkspaceClient(profile="MY_PROFILE")

---

コア API リファレンス

Clusters API

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/compute/clusters.html

# List all clusters
for cluster in w.clusters.list():
    print(f"{cluster.cluster_name}: {cluster.state}")

# Get cluster details
cluster = w.clusters.get(cluster_id="0123-456789-abcdef")

# Create a cluster (returns Wait object)
wait = w.clusters.create(
    cluster_name="my-cluster",
    spark_version=w.clusters.select_spark_version(latest=True),
    node_type_id=w.clusters.select_node_type(local_disk=True),
    num_workers=2
)
cluster = wait.result()  # Wait for cluster to be running

# Or use create_and_wait for blocking call
cluster = w.clusters.create_and_wait(
    cluster_name="my-cluster",
    spark_version="14.3.x-scala2.12",
    node_type_id="i3.xlarge",
    num_workers=2,
    timeout=timedelta(minutes=30)
)

# Start/stop/delete
w.clusters.start(cluster_id="...").result()
w.clusters.stop(cluster_id="...")
w.clusters.delete(cluster_id="...")

Jobs API

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/jobs/jobs.html

from databricks.sdk.service.jobs import Task, NotebookTask

# List jobs
for job in w.jobs.list():
    print(f"{job.job_id}: {job.settings.name}")

# Create a job
created = w.jobs.create(
    name="my-job",
    tasks=[
        Task(
            task_key="main",
            notebook_task=NotebookTask(notebook_path="/Users/me/notebook"),
            existing_cluster_id="0123-456789-abcdef"
        )
    ]
)

# Run a job now
run = w.jobs.run_now_and_wait(job_id=created.job_id)
print(f"Run completed: {run.state.result_state}")

# Get run output
output = w.jobs.get_run_output(run_id=run.run_id)

SQL ステートメントの実行

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/sql/statement_execution.html

# Execute SQL query
response = w.statement_execution.execute_statement(
    warehouse_id="abc123",
    statement="SELECT * FROM catalog.schema.table LIMIT 10",
    wait_timeout="30s"
)

# Check status and get results
if response.status.state == StatementState.SUCCEEDED:
    for row in response.result.data_array:
        print(row)

# For large results, fetch chunks
chunk = w.statement_execution.get_statement_result_chunk_n(
    statement_id=response.statement_id,
    chunk_index=0
)

SQL Warehouses

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/sql/warehouses.html

# List warehouses
for wh in w.warehouses.list():
    print(f"{wh.name}: {wh.state}")

# Get warehouse
warehouse = w.warehouses.get(id="abc123")

# Create warehouse
created = w.warehouses.create_and_wait(
    name="my-warehouse",
    cluster_size="Small",
    max_num_clusters=1,
    auto_stop_mins=15
)

# Start/stop
w.warehouses.start(id="abc123").result()
w.warehouses.stop(id="abc123").result()

Unity Catalog - テーブル

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/catalog/tables.html

# List tables in a schema
for table in w.tables.list(catalog_name="main", schema_name="default"):
    print(f"{table.full_name}: {table.table_type}")

# Get table info
table = w.tables.get(full_name="main.default.my_table")
print(f"Columns: {[c.name for c in table.columns]}")

# Check if table exists
exists = w.tables.exists(full_name="main.default.my_table")

Unity Catalog - カタログとスキーマ

ドキュメント（Catalogs）： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/catalog/catalogs.html ドキュメント（Schemas）： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/catalog/schemas.html

# List catalogs
for catalog in w.catalogs.list():
    print(catalog.name)

# Create catalog
w.catalogs.create(name="my_catalog", comment="Description")

# List schemas
for schema in w.schemas.list(catalog_name="main"):
    print(schema.name)

# Create schema
w.schemas.create(name="my_schema", catalog_name="main")

ボリューム

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/catalog/volumes.html

from databricks.sdk.service.catalog import VolumeType

# List volumes
for vol in w.volumes.list(catalog_name="main", schema_name="default"):
    print(f"{vol.full_name}: {vol.volume_type}")

# Create managed volume
w.volumes.create(
    catalog_name="main",
    schema_name="default",
    name="my_volume",
    volume_type=VolumeType.MANAGED
)

# Read volume info
vol = w.volumes.read(name="main.default.my_volume")

Files API

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/files/files.html

# Upload file to volume
w.files.upload(
    file_path="/Volumes/main/default/my_volume/data.csv",
    contents=open("local_file.csv", "rb")
)

# Download file
with w.files.download(file_path="/Volumes/main/default/my_volume/data.csv") as f:
    content = f.read()

# List directory contents
for entry in w.files.list_directory_contents("/Volumes/main/default/my_volume/"):
    print(f"{entry.name}: {entry.is_directory}")

# Upload/download with progress (parallel)
w.files.upload_from(
    file_path="/Volumes/main/default/my_volume/large.parquet",
    source_path="/local/path/large.parquet",
    use_parallel=True
)

w.files.download_to(
    file_path="/Volumes/main/default/my_volume/large.parquet",
    destination="/local/output/",
    use_parallel=True
)

Serving エンドポイント（モデルサービング）

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/serving/serving_endpoints.html

# List endpoints
for ep in w.serving_endpoints.list():
    print(f"{ep.name}: {ep.state}")

# Get endpoint
endpoint = w.serving_endpoints.get(name="my-endpoint")

# Query endpoint
response = w.serving_endpoints.query(
    name="my-endpoint",
    inputs={"prompt": "Hello, world!"}
)

# For chat/completions endpoints
response = w.serving_endpoints.query(
    name="my-chat-endpoint",
    messages=[{"role": "user", "content": "Hello!"}]
)

# Get OpenAI-compatible client
openai_client = w.serving_endpoints.get_open_ai_client()

ベクトル検索

ドキュメント（インデックス）： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/vectorsearch/vector_search_indexes.html ドキュメント（エンドポイント）： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/vectorsearch/vector_search_endpoints.html

# List vector search indexes
for idx in w.vector_search_indexes.list_indexes(endpoint_name="my-vs-endpoint"):
    print(idx.name)

# Query index
results = w.vector_search_indexes.query_index(
    index_name="main.default.my_index",
    columns=["id", "text", "embedding"],
    query_text="search query",
    num_results=10
)
for doc in results.result.data_array:
    print(doc)

パイプライン（Delta Live Tables）

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/pipelines/pipelines.html

# List pipelines
for pipeline in w.pipelines.list_pipelines():
    print(f"{pipeline.name}: {pipeline.state}")

# Get pipeline
pipeline = w.pipelines.get(pipeline_id="abc123")

# Start pipeline update
w.pipelines.start_update(pipeline_id="abc123")

# Stop pipeline
w.pipelines.stop_and_wait(pipeline_id="abc123")

シークレット

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/workspace/workspace/secrets.html

# List secret scopes
for scope in w.secrets.list_scopes():
    print(scope.name)

# Create scope
w.secrets.create_scope(scope="my-scope")

# Put secret
w.secrets.put_secret(scope="my-scope", key="api-key", string_value="secret123")

# Get secret (returns GetSecretResponse with value)
secret = w.secrets.get_secret(scope="my-scope", key="api-key")

# List secrets in scope (metadata only, not values)
for s in w.secrets.list_secrets(scope="my-scope"):
    print(s.key)

DBUtils

ドキュメント： https://databricks-sdk-py.readthedocs.io/en/latest/dbutils.html

# Access dbutils through WorkspaceClient
dbutils = w.dbutils

# File system operations
files = dbutils.fs.ls("/")
dbutils.fs.cp("dbfs:/source", "dbfs:/dest")
dbutils.fs.rm("dbfs:/path", recurse=True)

# Secrets (same as w.secrets but dbutils interface)
value = dbutils.secrets.get(scope="my-scope", key="my-key")

---

よくあるパターン

重要：非同期アプリケーション（FastAPI など）

Databricks SDK は完全に同期型です。 すべての呼び出しでスレッドがブロックされます。非同期アプリケーション（FastAPI、asyncio）では、イベントループをブロックしないよう、SDK の呼び出しを必ず asyncio.to_thread() でラップしてください。

import asyncio
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# WRONG - blocks the event loop
async def get_clusters_bad():
    return list(w.clusters.list())  # BLOCKS!

# CORRECT - runs in thread pool
async def get_clusters_good():
    return await asyncio.to_thread(lambda: list(w.clusters.list()))

# CORRECT - for simple calls
async def get_cluster(cluster_id: str):
    return await asyncio.to_thread(w.clusters.get, cluster_id)

# CORRECT - FastAPI endpoint
from fastapi import FastAPI
app = FastAPI()

@app.get("/clusters")
async def list_clusters():
    clusters = await asyncio.to_thread(lambda: list(w.clusters.list()))
    return [{"id": c.cluster_id, "name": c.cluster_name} for c in clusters]

@app.post("/query")
async def run_query(sql: str, warehouse_id: str):
    # Wrap the blocking SDK call
    response = await asyncio.to_thread(
        w.statement_execution.execute_statement,
        statement=sql,
        warehouse_id=warehouse_id,
        wait_timeout="30s"
    )
    return response.result.data_array

注： WorkspaceClient().config.host はネットワーク呼び出しではなく、設定を読み取るだけです。プロパティへのアクセスをラップする必要はありません。

---

長時間実行されるオペレーションの待機

from datetime import timedelta

# Pattern 1: Use *_and_wait methods
cluster = w.clusters.create_and_wait(
    cluster_name="test",
    spark_version="14.3.x-scala2.12",
    node_type_id="i3.xlarge",
    num_workers=2,
    timeout=timedelta(minutes=30)
)

# Pattern 2: Use Wait object
wait = w.clusters.create(...)
cluster = wait.result()  # Blocks until ready

# Pattern 3: Manual polling with callback
def progress(cluster):
    print(f"State: {cluster.state}")

cluster = w.clusters.wait_get_cluster_running(
    cluster_id="...",
    timeout=timedelta(minutes=30),
    callback=progress
)

ページネーション

# All list methods return iterators that handle pagination automatically
for job in w.jobs.list():  # Fetches all pages
    print(job.settings.name)

# For manual control
from databricks.sdk.service.jobs import ListJobsRequest
response = w.jobs.list(limit=10)
for job in response:
    print(job)

エラー処理

from databricks.sdk.errors import NotFound, PermissionDenied, ResourceAlreadyExists

try:
    cluster = w.clusters.get(cluster_id="invalid-id")
except NotFound:
    print("Cluster not found")
except PermissionDenied:
    print("Access denied")

---

判断に迷ったとき

メソッドについて確信が持てない場合は、次の手順を実行してください。

1. ドキュメントの URL パターンを確認します。

   • https://databricks-sdk-py.readthedocs.io/en/latest/workspace/{category}/{service}.html

2. 一般的なカテゴリ：

   • クラスター：/workspace/compute/clusters.html
   • ジョブ：/workspace/jobs/jobs.html
   • テーブル：/workspace/catalog/tables.html
   • ウェアハウス：/workspace/sql/warehouses.html
   • サービング：/workspace/serving/serving_endpoints.html

3. パラメーターや戻り値の型について案内する前に、取得して確認します。

---

クイックリファレンスリンク

API	ドキュメント URL
認証	https://databricks-sdk-py.readthedocs.io/en/latest/authentication.html
クラスター	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/compute/clusters.html
ジョブ	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/jobs/jobs.html
SQL ウェアハウス	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/sql/warehouses.html
ステートメント実行	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/sql/statement_execution.html
テーブル	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/catalog/tables.html
カタログ	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/catalog/catalogs.html
スキーマ	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/catalog/schemas.html
ボリューム	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/catalog/volumes.html
ファイル	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/files/files.html
サービングエンドポイント	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/serving/serving_endpoints.html
ベクトル検索	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/vectorsearch/vector_search_indexes.html
パイプライン	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/pipelines/pipelines.html
シークレット	https://databricks-sdk-py.readthedocs.io/en/latest/workspace/workspace/secrets.html
DBUtils	https://databricks-sdk-py.readthedocs.io/en/latest/dbutils.html

関連スキル

• databricks-core - プロファイルと認証の設定
• databricks-dabs - DABs を使用したリソースのデプロイ
• databricks-jobs - ジョブのオーケストレーションパターン
• databricks-unity-catalog - カタログガバナンス
• databricks-model-serving - サービングエンドポイントの管理
• databricks-vector-search - ベクトルインデックスの操作
• databricks-lakebase - オートスケーリングとブランチ機能を備えたマネージド PostgreSQL