Databricks SDK for Python

この記事では、 Databricks SDK for Pythonを使用して Databricks アカウント、ワークスペース、および関連リソースの操作を自動化する方法を学習します。 この記事は、Read The Docs にある Databricks SDK for Pythonドキュメントと、GitHub の Databricks SDK for Python リポジトリのコード サンプルを補足するものです。

注：

この機能はベータ版であり、本番運用で使用しても問題ありません。

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

始める前に

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

• Databricks ノートブック内から Databricks SDK for Python を使用するには、「Databricks ノートブックから Databricks SDK for Python を使用する」に進んでください。

• ローカルの開発マシンからDatabricks SDK for Pythonを使用するには、このセクションの手順を実行します。

Databricks SDK for Pythonの使用を開始する前に、開発マシンが以下の要件を満たしていることを確認してください。

• Databricks 認証 が構成されています。

• Python 3.8 以降がインストールされています。 Databricksを自動化する場合、 Databricksでは、ターゲットの Databricks コンピュート リソースにインストールされているバージョンと一致するPythonのメジャー バージョンとマイナー バージョンをインストールすることをお勧めしますDatabricks この記事の例では、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 または Potetry を使用する方法について説明します。

venv を使用して Python 仮想環境を作成する

1. ターミナルセットからPythonコードプロジェクトのルートディレクトリに、次のコマンドを実行します。 このコマンドは、仮想環境に Python 3.10 を使用するように venv に指示し、Python コード プロジェクトのルート ディレクトリ内の .venv という名前の非表示ディレクトリに仮想環境のサポート ファイルを作成します。

   # Linux and macOS
   python3.10 -m venv ./.venv
   
   # Windows
   python3.10 -m venv .\.venv
   

2. venv を使用して仮想環境をアクティブ化します。オペレーティングシステムとターミナルごとに使用すべき正しいコマンドについては、venvドキュメントを参照してください。例えば、 zsh を実行しているmacOSでは、以下のようになります。

   source ./.venv/bin/activate
   

   ターミナルプロンプトの直前に仮想環境の名前が括弧内に表示されると（例：.venv）、仮想環境がアクティブ化されたことが分かります。

   任意のタイミングで仮想環境を非アクティブ化するには、deactivate のコマンドを実行します。

   ターミナルプロンプトの直前に仮想環境の名前が括弧内に表示されなくなると、仮想環境が非アクティブ化されたことが分かります。

「 Databricks SDK for Python の概要」に進んでください。

Poetryで仮想環境を作る

1. Poetry をまだインストールしていない場合は、インストールします。

2. ターミナルセットから Python コード プロジェクトのルート ディレクトリに移動し、次のコマンドを実行して、Python コード プロジェクトを Poetry 用に初期化するように poetry に指示します。

   poetry init
   

3. 「Poetry」には、入力を求めるプロンプトがいくつか表示されます。 これらのプロンプトはいずれも、Databricks SDK for Python に固有のものではありません。 これらのプロンプトの詳細については、「 init」を参照してください。

4. プロンプトを完了すると、Poetry によって Python プロジェクトに pyproject.toml ファイルが追加されます。 pyproject.toml ファイルの詳細については、「pyproject.toml ファイル」を参照してください。

5. ターミナルを Python コード プロジェクトのルート ディレクトリに設定したまま、次のコマンドを実行します。 このコマンドは、 poetry pyproject.toml ファイルを読み取り、依存関係をインストールして解決し、依存関係をロックする poetry.lock ファイルを作成し、最後に仮想環境を作成するように指示します。

   poetry install
   

6. ターミナルセットからPythonコードプロジェクトのルートディレクトリまで、次のコマンドを実行して、仮想環境をアクティブ化してシェルに入るように poetry に指示します。

   poetry shell
   

   仮想環境がアクティブ化され、仮想環境の名前がターミナルプロンプトの直前に括弧内に表示されると、シェルが入力されたことがわかります。

   仮想環境を非アクティブ化してシェルを終了するには、コマンド exitを実行します。

   シェルを終了したことは、仮想環境の名前がターミナルプロンプトの直前に括弧内に表示されなくなったときです。

   Poetry 仮想環境の作成と管理の詳細については、「 環境の管理」を参照してください。

Databricks SDK for Pythonの使用を開始する

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

1. Databricks 認証が構成され、Python が既にインストールされ、Python 仮想環境が既にアクティブ化されている開発用コンピューターで、次のように Python Package Index (PyPI) から databricks-sdk パッケージ (およびその依存関係) をインストールします。

   pip を使用して databricks-sdk パッケージをインストールします。(一部のシステムでは、 pip3 を pipに置き換える必要がある場合があります。

   pip3 install databricks-sdk
   

   poetry add databricks-sdk
   

   Databricks SDK for Python がベータ版のときに特定のバージョンの databricks-sdk パッケージをインストールするには、パッケージの リリース履歴を参照してください。 たとえば、バージョン 0.1.6をインストールするには、次のようにします。

   pip3 install databricks-sdk==0.1.6
   

   poetry add databricks-sdk==0.1.6
   

   ヒント

   Databricks SDK for Python パッケージの既存のインストールを最新バージョンにアップグレードするには、次のコマンドを実行します。

   pip3 install --upgrade databricks-sdk
   

   poetry add databricks-sdk@latest
   

   Databricks SDK for Python パッケージの現在の Version とその他の詳細を表示するには、次のコマンドを実行します。

   pip3 show databricks-sdk
   

   poetry show databricks-sdk
   

2. Pythonの仮想環境で、Databricks SDK for PythonをインポートするPython コードファイルを作成します。次の例では、以下の内容を含む main.py という名前のファイル内に、Databricksワークスペース内のすべてのクラスターが一覧表示されています。

   from databricks.sdk import WorkspaceClient
   
   w = WorkspaceClient()
   
   for c in w.clusters.list():
     print(c.cluster_name)
   

3. python コマンドを実行して、ファイル名が main.py であると仮定して Pythonコードファイルを実行します。

   python3.10 main.py
   

   仮想環境のシェルにいる場合:

   python3.10 main.py
   

   仮想環境のシェルにいない場合は、次のようにします。

   poetry run python3.10 main.py
   

   注：

   前の w = WorkspaceClient()の呼び出しで引数を設定しないことで、Python 用 Databricks SDK は、既定のプロセスを使用して Databricks 認証を実行しようとします。 このデフォルト動作をオーバーライドするには、次の 認証 セクションを参照してください。

Databricksアカウントまたはワークスペースを使用してDatabricks SDK for Pythonを認証する

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

Databricks SDK for Python は、認証に対する統合された一貫性のあるアーキテクチャとプログラムによるアプローチである Databricks クライアント統合認証 標準を実装します。 このアプローチにより、Databricks での認証の設定と自動化が一元化され、予測可能になります。 これにより、Databricks 認証を一度構成すると、認証構成をさらに変更することなく、複数の Databricks ツールと SDK でその構成を使用できます。 Python のより完全なコード例など、詳細については、「 Databricks クライアント統合認証」を参照してください。

Databricks SDK for Pythonを使用してDatabricks認証を初期化するために使用できるコーディングのパターンには、以下のような例が挙げられます

• 以下のいずれかを実行して、Databricksのデフォルト認証を使用します。

  • ターゲットの Databricks 認証の種類に必要なフィールドを使用して、カスタム Databricks 構成プロファイル を作成または識別します。 次に、 DATABRICKS_CONFIG_PROFILE 環境変数をカスタム構成プロファイルの名前に設定します。

  • ターゲットのDatabricks認証タイプに必要とされる環境変数を設定します。

  次に、たとえば、次のように Databricks デフォルト認証を使用して WorkspaceClient オブジェクトをインスタンス化します。

  from databricks.sdk import WorkspaceClient
  
  w = WorkspaceClient()
  # ...
  

• 必須フィールドのハードコーディングはサポートされていますが、Databricks パーソナルアクセストークンなどのコード内の機密情報が公開される危険があるため推奨はされません。以下の例では、Databricksトークン認証用にDatabricksホストとアクセストークンの値がハードコーディングされています。

  from databricks.sdk import WorkspaceClient
  
  w = WorkspaceClient(
    host  = 'https://...',
    token = '...'
  )
  # ...
  

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

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

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

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

Databricks SDK for Python 0.6.0 以降では、デフォルトの Databricks ノートブック認証が使用されます。 デフォルトの Databricks ノートブック認証は、Databricks が独自に使用するためにバックグラウンドで自動的に生成する一時的な Databricks 個人アクセス サイトに依存します。 Databricks は、ノートブックの実行が停止した後、この一時トークンを削除します。

重要

デフォルト Databricks ノートブック認証は、クラスターのドライバー ノードでのみ機能し、クラスターのワーカー ノードまたはエグゼキューター ノードでは機能しません。

Databricks Runtime 13.3 LTS 以降では、Databricks SDK for Python 0.1.7 以降がインストールされている場合のデフォルトの Databricks ノートブック認証がサポートされます。 Databricks Runtime 10.4 LTS 以降では、Databricks SDK for Python 0.1.10 を使用したデフォルトの Databricks ノートブック認証がサポートされています。 以上がインストールされています。 ただし、Databricks DatabricksSDKPythonDatabricksでは、 バージョンに関係なく、 デフォルト ユーザー名認証との互換性を最大限に高めるために、 for 0.6.0Databricks Runtime 以降をインストールするか、これにアップグレードすることをお勧めします。

Databricks アカウント レベルのAPIsを呼び出す場合、またはデフォルトの Databricks ノートブック認証以外の Databricks 認証タイプを使用する場合は、次のように Databricks クラスターで Databricks SDK for Python をインストールまたはアップグレードする必要があります。

認証タイプ	Databricks SDK for Python のバージョン
OAuth マシン間 (M2M) 認証	すべてのバージョン
OAuth ユーザー対マシン (U2M) 認証	0.1.9 以上
Databricks の個人アクセス時の認証	すべてのバージョン
基本認証 (レガシー) (本番運用では推奨されません)	すべてのバージョン

Databricks ノートブック認証は、Databricks 構成プロファイルでは機能しません。

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

1. Databricks Python ノートブックでは、他の Python ライブラリと同様に、Databricks SDK for Python を使用できます。 接続された Databricks クラスターに Databricks SDK for Python ライブラリをインストールまたはアップグレードするには、次のようにノートブック セルから%pipマジック コマンドを実行します。

   %pip install databricks-sdk --upgrade
   

2. %pipマジック コマンドを実行した後、Python を再起動して、インストールまたはアップグレードされたライブラリをノートブックで使用できるようにする必要があります。 これを行うには、ノートブックのセルのセルの直後に、 %pipマジック コマンドを使用して次のコマンドを実行します。

   dbutils.library.restartPython()
   

3. Databricks SDK for Python のインストールされているバージョンを表示するには、ノートブックのセルから次のコマンドを実行します。

   %pip show databricks-sdk | grep -oP '(?<=Version: )\S+'
   

ステップ 2: コードを実行する

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

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

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

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

別の Databricks 認証タイプを使用する場合は、 「サポートされている Databricks 認証タイプ」を参照し、対応するリンクをクリックして追加の技術的な詳細を確認してください。

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

Databricks ユーティリティ (dbutils) リファレンスは、ローカル開発コンピューターで実行されている Python コード用の Databricks SDK から、または Databricks ノートブック内から呼び出すことができます。

• ローカル開発マシンから、Databricks ユーティリティは dbutils.fs、 dbutils.secrets、 dbutils.widgets、 dbutils.jobs コマンド グループにのみアクセスできます。

• Databricksクラスターに接続されているDatabricksノートブックから、Databricksユーティリティは、dbutils.fs 、 dbutils.secrets 、および dbutils.widgets だけでなく、使用可能なすべてのDatabricksユーティリティのコマンドグループにアクセスできます。さらに、 dbutils.notebook コマンドグループは2つのレベルのコマンドのみに制限されます。（例：dbutils.notebook.run または dbutils.notebook.exit ）

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

from databricks.sdk import WorkspaceClient

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

for f in d:
  print(f.path)

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

from databricks.sdk.runtime import *

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

for f in d:
  print(f.path)

dbutils を単独で使用することも、WorkspaceClient内で使用して Unity Catalog ボリュームにアクセスすることもできません。代わりに、WorkspaceClient内でfilesを使用します。このコード例では、WorkspaceClient内のfilesを呼び出して、指定したボリューム内の指定したファイルの内容を出力します。

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

resp = w.files.download('/Volumes/main/default/my-volume/sales.csv')
print(str(resp.contents.read(), encoding='utf-8'))

dbutilsの操作も参照してください。

コード例

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

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

• DatabricksワークスペースのAPI リファレンス

• DatabricksアカウントのAPIリファレンス

クラスターを作成する

このコード例では、指定された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             = 'i3.xlarge',
  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)

ジョブを作成する

このコード例では、指定されたクラスター上で特定のノートブックを実行する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")

サーバーレスコンピュートを使ったジョブを作成する

プレビュー

サーバレス コンピュート for ワークフローはパブリック プレビュー中です。 資格と有効化に関する情報については、 「サーバレス コンピュートのパブリック プレビューを有効にする」を参照してください。

次の例では 、ワークフローにサーバレス コンピュートを使用するジョブを作成します。

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",
   )
  ]
)

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

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

from databricks.sdk import AccountClient

a = AccountClient()

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

テスティング

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

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

# 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ワークスペースのAPI リファレンス

• DatabricksアカウントのAPIリファレンス

• ロギング

• 長期にわたるオペレーション

• ページ分割されたレスポンス

• OAuthによるシングルサインオン