Databricks Python SDK

pip install databricks-sdk

from databricks.sdk import WorkspaceClient

# Databricks CLI 로그인이 되어 있으면 자동으로 인증
w = WorkspaceClient()

# 현재 사용자 확인
me = w.current_user.me()
print(f"인증된 사용자: {me.user_name}")

from databricks.sdk import WorkspaceClient

w = WorkspaceClient(
    host="https://<workspace-url>",
    token="dapi..."
)

from databricks.sdk import WorkspaceClient

w = WorkspaceClient(
    host="https://<workspace-url>",
    client_id="<client-id>",
    client_secret="<client-secret>"
)

export DATABRICKS_HOST="https://<workspace-url>"
export DATABRICKS_TOKEN="dapi..."

# 환경 변수가 설정되어 있으면 자동으로 인식
w = WorkspaceClient()

from databricks.sdk import WorkspaceClient, AccountClient

# 워크스페이스 클라이언트 (가장 많이 사용)
w = WorkspaceClient()

# 계정 클라이언트 (관리자용)
a = AccountClient(
    host="https://accounts.cloud.databricks.com",
    account_id="<account-id>"
)

# 클러스터 목록 조회
for cluster in w.clusters.list():
    print(f"{cluster.cluster_name}: {cluster.state}")

# 클러스터 시작 (완료까지 대기)
w.clusters.start(cluster_id="0123-456789-abcdef").result()
print("클러스터가 시작되었습니다.")

# Job 목록 조회
for job in w.jobs.list():
    print(f"[{job.job_id}] {job.settings.name}")

# Job 즉시 실행 및 완료 대기
run = w.jobs.run_now(job_id=12345)
result = run.result()
print(f"실행 완료: {result.state.result_state}")

# SQL 쿼리 실행
from databricks.sdk.service.sql import StatementState

response = w.statement_execution.execute_statement(
    warehouse_id="abc123def456",
    statement="SELECT region, SUM(revenue) AS total FROM sales GROUP BY region ORDER BY total DESC LIMIT 10",
    wait_timeout="30s"
)

if response.status.state == StatementState.SUCCEEDED:
    # 컬럼명 출력
    columns = [col.name for col in response.manifest.schema.columns]
    print("\t".join(columns))

    # 결과 행 출력
    for row in response.result.data_array:
        print("\t".join(str(v) for v in row))

# 3-level 네임스페이스 탐색: 카탈로그 → 스키마 → 테이블
for catalog in w.catalogs.list():
    print(f"카탈로그: {catalog.name}")

for table in w.tables.list(catalog_name="my_catalog", schema_name="my_schema"):
    print(f"  테이블: {table.name} ({table.table_type})")

# 권한 부여
from databricks.sdk.service.catalog import SecurableType, PermissionsChange, Privilege

w.grants.update(
    securable_type=SecurableType.TABLE,
    full_name="my_catalog.my_schema.customers",
    changes=[PermissionsChange(add=[Privilege.SELECT], principal="data-analysts")]
)

# 엔드포인트 호출
response = w.serving_endpoints.query(
    name="my-model-endpoint",
    dataframe_records=[{"feature1": 1.0, "feature2": "A", "feature3": 100}]
)
print(f"예측 결과: {response.predictions}")

# Secret Scope 생성 및 Secret 저장
w.secrets.create_scope(scope="my-app-secrets")
w.secrets.put_secret(scope="my-app-secrets", key="api-key", string_value="sk-...")

pip install databricks-connect==15.4.*

from databricks.connect import DatabricksSession

# 원격 Spark 세션 생성
spark = DatabricksSession.builder.remote(
    host="https://<workspace-url>",
    cluster_id="0123-456789-abcdef"
).getOrCreate()

# 로컬 코드가 원격 클러스터에서 실행됨
df = spark.table("my_catalog.my_schema.orders")
result = df.groupBy("region").sum("revenue").collect()

for row in result:
    print(f"{row['region']}: {row['sum(revenue)']}")

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# 현재 사용자 확인
me = w.current_user.me()
print(f"사용자: {me.user_name}")

# 클러스터 현황
clusters = list(w.clusters.list())
running = [c for c in clusters if str(c.state) == "RUNNING"]
print(f"클러스터: 총 {len(clusters)}개, 실행 중 {len(running)}개")

# Job / Warehouse / 서빙 엔드포인트 현황
print(f"Job: 총 {len(list(w.jobs.list()))}개")
print(f"SQL Warehouse: 총 {len(list(w.warehouses.list()))}개")
print(f"서빙 엔드포인트: 총 {len(list(w.serving_endpoints.list()))}개")