Databricks SDK for Python

[アーティクル]
10/11/2024

Note

Databricks では、ジョブやその他の Databricks リソースをソースコードとして作成、開発、デプロイ、テストするための Databricks アセットバンドルをお勧めします。「Databricks アセットバンドルとは」をご覧ください。

この記事では、 Databricks SDK for Python を使用して Azure Databricks 操作を自動化し、開発を高速化する方法について説明します。この記事は、参考資料の Databricks SDK for Python のドキュメントと、GitHub の Databricks SDK for Python リポジトリにあるコードサンプルを補足するものです。

Note

Databricks SDK for Python は Beta にあり、運用環境で使用しても問題ありません。

ベータ期間中、Databricks では、コードが依存する Databricks SDK for Python の特定のマイナーバージョンへの依存関係をピン留めすることをお勧めします。たとえば、 venv の requirements.txtや Poetry の pyproject.toml と poetry.lock などのファイルに依存関係をピン留めできます。依存関係のピン留めについて詳しくは、venv で「仮想環境とパッケージ」または Poetryに「依存関係をインストールする」を参照してください。

開始する前に

Databricks SDK for Python は、Azure Databricks ノートブック内かローカル開発用マシンから使用できます。

Azure Databricks ノートブック内から Databricks SDK for Python を使用する方法を確認するには、「Azure Databricks ノートブックから Databricks SDK for Python を使用する」に進んでください。
ローカル開発用マシンから Databricks SDK for Python を使用するには、このセクションで示す手順を実行してください。

Databricks SDK for Python の使用を開始する前に、開発用マシンで次の条件が満たされている必要があります。

Azure Databricks 認証が構成されている。
Python 3.8 以上がインストールされている。 Azure Databricks コンピューティングリソースを自動化する場合は、Databricks で対象の Azure Databricks コンピューティングリソースにインストールされているバージョンと一致する Python メジャーバージョンおよびマイナーバージョンをインストールすることをお勧めします。この記事の例では、Python 3.10 がインストールされている Databricks Runtime 13.3 LTS を使用するクラスターの自動化を利用しています。ご使用のクラスターの Databricks Runtime について正しいバージョンを確認するには、「Databricks Runtime リリースノートのバージョンと互換性」をご覧ください。
Databricks では、Databricks SDK for Python で使用する Python プロジェクトごとに Python 仮想環境を作成しアクティブ化することをおすすめしています。 Python 仮想環境は、コードプロジェクトで必ず互換バージョンの Python および Python パッケージ (この場合は、Databricks SDK for Python パッケージ) が使用されるようにするために役立ちます。 Python 仮想環境の詳細については、venv またはポエムを参照してください。

Databricks SDK for Python での作業開始

このセクションでは、ローカル開発用マシンから Databricks SDK for Python の使用を開始する方法について説明します。 Azure Databricks ノートブック内から Databricks SDK for Python を使用する方法を確認するには、「Azure Databricks ノートブックから Databricks SDK for Python を使用する」に進んでください。

Azure Databricks 認証が構成され、Python が既にインストールされ、Python 仮想環境が既にアクティブ化されている開発マシンに、次のように Python パッケージインデックス (PyPI) から databricks-sdk パッケージ (およびその依存関係) をインストールします:

Venv

pip を使用して、databricks-sdk パッケージをインストールします。 (一部のシステムでは、ここと全体で、pip3 を pip に置き換える必要がある場合があります。)
```
pip3 install databricks-sdk
```
Poetry
```
poetry add databricks-sdk
```
Databricks SDK for Python がベータ版である間に特定のバージョンの databricks-sdk パッケージをインストールするには、パッケージのリリース履歴を参照してください。たとえば、バージョン 0.1.6 をインストールする場合:

Venv
```
pip3 install databricks-sdk==0.1.6
```
Poetry
```
poetry add databricks-sdk==0.1.6
```
Databricks SDK for Python パッケージの既存のインストールを最新バージョンにアップグレードするには、次のコマンドを実行します:

Venv
```
pip3 install --upgrade databricks-sdk
```
Poetry
```
poetry add databricks-sdk@latest
```
Databricks SDK for Python パッケージの現在の Version とその他の詳細を表示するには、次のコマンドを実行します:

Venv
```
pip3 show databricks-sdk
```
Poetry
```
poetry show databricks-sdk
```
Python 仮想環境で、Databricks SDK for Python をインポートする Python コードファイルを作成します。次の例では、次のコンテンツの main.py という名前のファイルで、Azure Databricks ワークスペース内のすべてのクラスターを一覧表示しているだけです。
```
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

for c in w.clusters.list():
  print(c.cluster_name)
```
次のように、python コマンドを実行して、main.py という名前のファイルを想定し Python コードファイルを実行します。

Venv
```
python3.10 main.py
```
Poetry

仮想環境のシェル内にある場合:
```
python3.10 main.py
```
仮想環境のシェルにない場合:
```
poetry run python3.10 main.py
```
Note

前述の w = WorkspaceClient() の呼び出しで引数を設定しないと、Databricks SDK for Python で、Azure Databricks 認証の実行を試みるためにその既定のプロセスが使用されます。この既定の動作をオーバーライドする方法については、認証に関する次のセクションをご覧ください。

Azure Databricks アカウントまたはワークスペースで Databricks SDK for Python を認証する

このセクションでは、Azure Databricks アカウントまたはワークスペースによってローカル開発用マシンから Databricks SDK for Python を認証する方法を説明します。 Azure Databricks ノートブック内から Databricks SDK for Python を認証する方法を確認するには、「Azure Databricks ノートブックから Databricks SDK for Python を使用する」に進んでください。

Databricks SDK for Python では、Databricks クライアント統合認証標準が実装されます。これは、統合された、一貫性のある、アーキテクチャとプログラムによる認証手法です。このアプローチは、Azure Databricks を使用した認証の設定と自動化を、より一元的で予測可能なものにするのに役立ちます。これにより、Databricks 認証を一度構成すれば、それ以上認証構成を変更しなくても、複数の Databricks ツールおよび SDK でその構成を使用できます。詳しくは (より完成された Python コード例など)、Databricks クライアント統合認証に関するページをご覧ください。

Note

Databricks SDK for Python では、まだ Azure マネージド ID 認証が実装されていません。

Databricks SDK for Python を使用して Databricks 認証を初期化するために使用できるコーディングパターンをいくつか次に示します。

次のいずれかを行って、Databricks の既定の認証を使用します。
- ターゲットの Databricks 認証の種類に必要なフィールドを含むカスタムの Databricks 構成プロファイルを作成または識別します。次に、DATABRICKS_CONFIG_PROFILE 環境変数をカスタム構成プロファイルの名前に設定します。
- ターゲットの Databricks 認証の種類に必要な環境変数を設定します。
次に、たとえば、以下のように Databricks の既定の認証を使って WorkspaceClient オブジェクトのインスタンスを作成します。
```
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
# ...
```
必要なフィールドのハードコーディングはサポートされていますが、Azure Databricks 個人用アクセストークンなどの機密情報がコードで公開されるリスクがあるため、推奨されません。次の例では、Databricks トークン認証用に Azure Databricks ホストとアクセストークンの値をハードコードします。
```
from databricks.sdk import WorkspaceClient

w = WorkspaceClient(
  host  = 'https://...',
  token = '...'
)
# ...
```

Databricks SDK for Python ドキュメントの「Authentication」(認証) も参照してください。

Azure Databricks ノートブックから Databricks SDK for Python を使用する

Azure Databricks クラスターがアタッチされ、Databricks SDK for Python がインストールされている Azure Databricks ノートブックから、Databricks SDK for Python の機能を呼び出すことができます。これは、Databricks Runtime 13.3 LTS 以上を使用するすべての Azure Databricks クラスターに既定でインストールされています。 Databricks Runtime 12.2 LTS 以下を使用する Azure Databricks クラスターの場合、まず Databricks SDK for Python をインストールする必要があります。「手順 1: Databricks SDK for Python をインストールまたはアップグレードする」を参照してください。

特定の Databricks Runtime バージョンにインストールされている Databricks SDK for Python のバージョンを確認するには、その Databricks Runtime バージョンの「Databricks Runtime リリースノート」の「インストール済みの Python ライブラリ」セクションを参照してください。

Databricks では、 PiPy から入手可能な最新バージョンの SDK のインストールをおすすめしますが、既定の Azure Databricks ノートブック認証がバージョン 0.6.0 以降のすべての Databricks Runtime バージョンで使用されているため、最低でも、Databricks SDK for Python 0.6.0 以降をインストールまたはアップグレードしてください。

Note

Databricks Runtime 15.1 は、アップグレード不要で、既定のノートブック認証をサポートする Databricks SDK for Python (0.20.0) のバージョンがインストールされた初の Databricks Runtime です。

次の表は、Databricks SDK for Python と Databricks Runtime のバージョンにおけるノートブック認証のサポート概要を示しています。

SDK/DBR	10.4 LTS	11.3 LTS	12.3 LTS	13.3 LTS	14.3 LTS	15.1 以降
0.1.7 以前
0.1.10		✓	✓	✓	✓	✓
0.6.0	✓	✓	✓	✓	✓	✓
0.20.0 以降	✓	✓	✓	✓	✓	✓

既定の Azure Databricks ノートブック認証は、Azure Databricks が独自に使用するためにバックグラウンドで自動的に生成する一時的な Azure Databricks 個人用アクセストークンに依存します。この一時トークンは、ノートブックの実行が停止すると、Azure Databricks によって削除されます。

重要

既定の Azure Databricks ノートブック認証が機能するのは、クラスターのドライバーノード上のみであり、クラスターのワーカーノードや Executor ノード上では機能しません。
Azure Databricks ノートブック認証は、Azure Databricks 構成プロファイルでは機能しません。

Azure Databricks account-level API を呼び出す場合、または既定の Databricks ノートブック認証以外の Databricks 認証の種類を使用する場合は、次の認証の種類もサポートされます。

認証の種類	Databricks SDK for Python のバージョン
OAuth マシン間 (M2M) 認証	0.18.0 以降
OAuth ユーザー対マシン (U2M) 認証	0.19.0 以降
Microsoft Entra ID サービスプリンシパル認証	すべてのバージョン
Azure CLI 認証	すべてのバージョン
Databricks 個人用アクセストークン認証	すべてのバージョン

Azure マネージド ID 認証はまだサポートされていません。

手順 1: Databricks SDK for Python をインストールまたはアップグレードする

Azure Databricks Python ノートブックでは、他の Python ライブラリと同様に Databricks SDK for Python を使用できます。アタッチされた Azure Databricks クラスターに Databricks SDK for Python ライブラリをインストールまたはアップグレードするには、次のようにノートブックセルから %pip マジックコマンドを実行します。
```
%pip install databricks-sdk --upgrade
```
%pip マジックコマンドを実行した後、Python を再起動して、インストールまたはアップグレードされたライブラリをノートブックで使用できるようにする必要があります。これを行うには、%pip マジックコマンドがあるセルの直後のノートブックセルから次のコマンドを実行します。
```
dbutils.library.restartPython()
```
Databricks SDK for Python のインストールされているバージョンを表示するには、ノートブックセルから次のコマンドを実行します。
```
%pip show databricks-sdk | grep -oP '(?<=Version: )\S+'
```

手順 2: コードを実行する

ノートブックセルで、Databricks SDK for Python をインポートしてから呼び出す、Python コードを作成します。次の例では、既定の Azure Databricks ノートブック認証を使用して、Azure Databricks ワークスペース内のすべてのクラスターを一覧表示します。

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

for c in w.clusters.list():
  print(c.cluster_name)

このセルを実行すると、Azure Databricks ワークスペース内の使用可能なすべてのクラスターの名前が一覧で表示されます。

別の種類の Azure Databricks 認証を使用するには、「Azure Databricks 認証の方法」を参照し、対応するリンクをクリックして技術的な詳細を確認してください。

Databricks ユーティリティを使用する

「Databricks Utilities (dbutils) リファレンス」は、ローカル開発用マシンで実行されている Databricks SDK for Python コードから、または Azure Databricks ノートブック内から呼び出すことができます。

ローカル開発用マシンからの場合、Databricks ユーティリティは、dbutils.fs、dbutils.secrets、dbutils.widgets、dbutils.jobs コマンドグループにのみアクセスできます。
Azure Databricks クラスターに関連付けられている Azure Databricks ノートブックからの場合、Databricks ユーティリティは、dbutils.fs、dbutils.secrets、dbutils.widgets のみでなく、使用可能なすべての Databricks ユーティリティコマンドグループにアクセスできます。また、dbutils.notebook コマンドグループは、dbutils.notebook.run や dbutils.notebook.exit など、2 つのレベルのコマンドのみに制限されます。

ローカル開発用マシンまたは Azure Databricks ノートブックから Databricks ユーティリティを呼び出すには、WorkspaceClient 内で dbutils を使用します。このコード例では、既定の Azure Databricks ノートブック認証を使用して、WorkspaceClient 内の dbutils を呼び出して、ワークスペースの DBFS ルート内のすべてのオブジェクトのパスを一覧表示します。

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
d = w.dbutils.fs.ls('/')

for f in d:
  print(f.path)

または、dbutils を直接呼び出すこともできます。ただし、既定の Azure Databricks ノートブック認証のみを使用するように制限されます。このコード例では、dbutils を直接呼び出して、ワークスペースの DBFS ルート内のすべてのオブジェクトを一覧表示します。

from databricks.sdk.runtime import *

d = dbutils.fs.ls('/')

for f in d:
  print(f.path)

Unity Catalog ボリュームにアクセスするには、WorkspaceClient 内の files を使用します。「Unity Catalog ボリューム内のファイルを管理する」を参照してください。ボリュームにアクセスするために、dbutils を単独でまたは WorkspaceClient 内で使用することはできません。

「dbutils とのやり取り」もご確認ください。

コード例

次のコード例では、Databricks SDK for Python を使用してクラスターを作成および削除し、ジョブを作成し、アカウントレベルのグループを一覧表示する方法を示します。これらのコード例では、既定の Azure Databricks ノートブック認証を使用します。既定の Azure Databricks ノートブック認証について詳しくは、「Azure Databricks ノートブック内から Databricks SDK for Python を使用する」をご覧ください。ノートブック以外の既定の認証について詳しくは、「Azure Databricks アカウントまたはワークスペースで Databricks SDK for Python を認証する」をご覧ください。

その他のコード例については、GitHub で Databricks SDK for Python リポジトリ内の例を参照してください。関連項目:

Databricks ワークスペース API リファレンス
Databricks アカウント API リファレンス
クラスターの作成
クラスターを完全に削除する
ジョブを作成する
サーバーレスコンピューティングを使用するジョブを作成する
Unity Catalog ボリューム内のファイルを管理する
アカウントレベルのグループを一覧表示する

クラスターの作成

このコード例では、Databricks Runtime バージョンとクラスターノードの種類を指定してクラスターを作成します。このクラスターには 1 つのワーカーがあり、クラスターは 15 分のアイドル時間の後に自動的に終了します。

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

print("Attempting to create cluster. Please wait...")

c = w.clusters.create_and_wait(
  cluster_name             = 'my-cluster',
  spark_version            = '12.2.x-scala2.12',
  node_type_id             = 'Standard_DS3_v2',
  autotermination_minutes  = 15,
  num_workers              = 1
)

print(f"The cluster is now ready at " \
      f"{w.config.host}#setting/clusters/{c.cluster_id}/configuration\n")

クラスターを完全に削除する

このコード例では、指定したクラスター ID のクラスターをワークスペースから完全に削除しています。

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

c_id = input('ID of cluster to delete (for example, 1234-567890-ab123cd4): ')

w.clusters.permanent_delete(cluster_id = c_id)

ジョブの作成

このコード例では、指定したクラスターで指定したノートブックを実行する Azure Databricks ジョブを作成しています。コードを実行すると、ターミナルでユーザーから既存のノートブックのパス、既存のクラスター ID、および関連するジョブ設定が取得されます。

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

w = WorkspaceClient()

job_name            = input("Some short name for the job (for example, my-job): ")
description         = input("Some short description for the job (for example, My job): ")
existing_cluster_id = input("ID of the existing cluster in the workspace to run the job on (for example, 1234-567890-ab123cd4): ")
notebook_path       = input("Workspace path of the notebook to run (for example, /Users/someone@example.com/my-notebook): ")
task_key            = input("Some key to apply to the job's tasks (for example, my-key): ")

print("Attempting to create the job. Please wait...\n")

j = w.jobs.create(
  name = job_name,
  tasks = [
    Task(
      description = description,
      existing_cluster_id = existing_cluster_id,
      notebook_task = NotebookTask(
        base_parameters = dict(""),
        notebook_path = notebook_path,
        source = Source("WORKSPACE")
      ),
      task_key = task_key
    )
  ]
)

print(f"View the job at {w.config.host}/#job/{j.job_id}\n")

サーバーレスコンピューティングを使用するジョブを作成する

次の例では、ジョブにサーバーレスコンピューティングを使用するジョブを作成します。

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

w = WorkspaceClient()

j = w.jobs.create(
  name = "My Serverless Job",
  tasks = [
    Task(
      notebook_task = NotebookTask(
      notebook_path = "/Users/user@databricks.com/MyNotebook",
      source = Source("WORKSPACE")
      ),
      task_key = "MyTask",
   )
  ]
)

Unity Catalog ボリューム内のファイルを管理する

このコード例では、Unity Catalog ボリュームにアクセスするための、WorkspaceClient 内の files 機能へのさまざまな呼び出しを示します。

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Define volume, folder, and file details.
catalog            = 'main'
schema             = 'default'
volume             = 'my-volume'
volume_path        = f"/Volumes/{catalog}/{schema}/{volume}" # /Volumes/main/default/my-volume
volume_folder      = 'my-folder'
volume_folder_path = f"{volume_path}/{volume_folder}" # /Volumes/main/default/my-volume/my-folder
volume_file        = 'data.csv'
volume_file_path   = f"{volume_folder_path}/{volume_file}" # /Volumes/main/default/my-volume/my-folder/data.csv
upload_file_path   = './data.csv'

# Create an empty folder in a volume.
w.files.create_directory(volume_folder_path)

# Upload a file to a volume.
with open(upload_file_path, 'rb') as file:
  file_bytes = file.read()
  binary_data = io.BytesIO(file_bytes)
  w.files.upload(volume_file_path, binary_data, overwrite = True)

# List the contents of a volume.
for item in w.files.list_directory_contents(volume_path):
  print(item.path)

# List the contents of a folder in a volume.
for item in w.files.list_directory_contents(volume_folder_path):
  print(item.path)

# Print the contents of a file in a volume.
resp = w.files.download(volume_file_path)
print(str(resp.contents.read(), encoding='utf-8'))

# Delete a file from a volume.
w.files.delete(volume_file_path)

# Delete a folder from a volume.
w.files.delete_directory(volume_folder_path)

アカウントレベルのグループを一覧表示する

このコード例では、Azure Databricks アカウント内で使用可能なすべてのグループの表示名を一覧表示します。

from databricks.sdk import AccountClient

a = AccountClient()

for g in a.groups.list():
  print(g.display_name)

テスト

コードをテストするには、pytest などの Python テストフレームワークを使用します。 Azure Databricks REST API エンドポイントを呼び出したり、Azure Databricks アカウントまたはワークスペースの状態を変更したりせずに、シミュレートされた条件下でコードをテストするには、unittest.mock などの Python モックライブラリを使用します。

ヒント

Databricks Labs には、 pytest プラグイン Databricks と pylint プラグインとの統合テストを簡略化しコードの品質を確保する機能が用意されています。

helpers.pyという名前の次のファイル例には、新しいクラスターに関する情報を返すcreate_cluster関数が含まれています。

# helpers.py

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.compute import ClusterDetails

def create_cluster(
  w: WorkspaceClient,
  cluster_name:            str,
  spark_version:           str,
  node_type_id:            str,
  autotermination_minutes: int,
  num_workers:             int
) -> ClusterDetails:
  response = w.clusters.create(
    cluster_name            = cluster_name,
    spark_version           = spark_version,
    node_type_id            = node_type_id,
    autotermination_minutes = autotermination_minutes,
    num_workers             = num_workers
  )
  return response

create_cluster関数を呼び出すmain.pyという名前の次のファイルを指定します。

# main.py

from databricks.sdk import WorkspaceClient
from helpers import *

w = WorkspaceClient()

# Replace <spark-version> with the target Spark version string.
# Replace <node-type-id> with the target node type string.
response = create_cluster(
  w = w,
  cluster_name            = 'Test Cluster',
  spark_version           = '<spark-version>',
  node_type_id            = '<node-type-id>',
  autotermination_minutes = 15,
  num_workers             = 1
)

print(response.cluster_id)

test_helpers.py という名前の次のファイルは、create_cluster 関数が、想定される応答を返すかどうかをテストします。このテストでは、ターゲットワークスペースにクラスターを作成するのではなく、WorkspaceClient オブジェクトをモックし、モックオブジェクトの設定を定義し、モックオブジェクトを create_cluster 関数に渡します。このテストでは次に、関数が、新しいモッククラスターの想定される ID を返すかどうかを確認します。

# test_helpers.py

from databricks.sdk import WorkspaceClient
from helpers import *
from unittest.mock import create_autospec # Included with the Python standard library.

def test_create_cluster():
  # Create a mock WorkspaceClient.
  mock_workspace_client = create_autospec(WorkspaceClient)

  # Set the mock WorkspaceClient's clusters.create().cluster_id value.
  mock_workspace_client.clusters.create.return_value.cluster_id = '123abc'

  # Call the actual function but with the mock WorkspaceClient.
  # Replace <spark-version> with the target Spark version string.
  # Replace <node-type-id> with the target node type string.
  response = create_cluster(
    w = mock_workspace_client,
    cluster_name            = 'Test Cluster',
    spark_version           = '<spark-version>',
    node_type_id            = '<node-type-id>',
    autotermination_minutes = 15,
    num_workers             = 1
  )

  # Assert that the function returned the mocked cluster ID.
  assert response.cluster_id == '123abc'

このテストを実行するには、コードプロジェクトのルートから pytest コマンドを実行します。これにより、次のようなテスト結果が生成されるはずです。

$ pytest
=================== test session starts ====================
platform darwin -- Python 3.12.2, pytest-8.1.1, pluggy-1.4.0
rootdir: <project-rootdir>
collected 1 item

test_helpers.py . [100%]
======================== 1 passed ==========================

その他のリソース

詳細については、以下を参照してください:

次の方法で共有

Databricks SDK for Python

開始する前に

Databricks SDK for Python での作業開始

Venv

Poetry

Venv

Poetry

Venv

Poetry

Venv

Poetry

Venv

Poetry

Azure Databricks アカウントまたはワークスペースで Databricks SDK for Python を認証する

Azure Databricks ノートブックから Databricks SDK for Python を使用する

手順 1: Databricks SDK for Python をインストールまたはアップグレードする

手順 2: コードを実行する

Databricks ユーティリティを使用する

コード例

クラスターの作成

クラスターを完全に削除する

ジョブの作成

サーバーレスコンピューティングを使用するジョブを作成する

Unity Catalog ボリューム内のファイルを管理する

アカウントレベルのグループを一覧表示する

テスト

その他のリソース

フィードバック

その他のリソース

次の方法で共有

Databricks SDK for Python

開始する前に

Databricks SDK for Python での作業開始

Venv

Poetry

Venv

Poetry

Venv

Poetry

Venv

Poetry

Venv

Poetry

Azure Databricks アカウントまたはワークスペースで Databricks SDK for Python を認証する

Azure Databricks ノートブックから Databricks SDK for Python を使用する

手順 1: Databricks SDK for Python をインストールまたはアップグレードする

手順 2: コードを実行する

Databricks ユーティリティを使用する

コード例

クラスターの作成

クラスターを完全に削除する

ジョブの作成

サーバーレス コンピューティングを使用するジョブを作成する

Unity Catalog ボリューム内のファイルを管理する

アカウントレベルのグループを一覧表示する

テスト

その他のリソース

フィードバック

その他のリソース

サーバーレスコンピューティングを使用するジョブを作成する