Python でのバンドル構成

Python for Databricks アセットバンドルは Databricks アセットバンドルを拡張して、次のことを可能にします。

Python コードでリソースを定義します。これらの定義は、 YAML で定義されたリソースと共存できます。
メタデータを使用してリソースを動的に作成します。メタデータを使用したリソースの作成を参照してください。
バンドルのデプロイ中に YAML または Python で定義されているリソースを変更します。 YAML または Python で定義されているリソースの変更に関するページを参照してください。

Databricks Asset Bundles databricks-bundles パッケージの Python サポートに関するリファレンスドキュメントは、 https://databricks.github.io/cli/python/で入手できます。

要求事項

Databricks アセットバンドルの Python サポートを使用するには、まず次の手順を実行する必要があります。

Databricks CLI バージョン 0.275.0 以降をインストールします。 Databricks CLI のインストールまたは更新を参照してください。
まだ認証していない場合は、Databricks ワークスペースに対して認証します。
```
databricks configure
```
uv をインストールします。 uv の取り付けを参照してください。 Python for Databricks Asset Bundles では、uv を使用して仮想環境を作成し、必要な依存関係をインストールします。または、 venv などの他のツールを使用して Python 環境を構成することもできます。

テンプレートからプロジェクトを作成する

Databricks Asset Bundles プロジェクトの新しい Python サポートを作成するには、 pydabs テンプレートを使用してバンドルを初期化します。

databricks bundle init pydabs

メッセージが表示されたら、プロジェクトに my_pydabs_projectなどの名前を付け、ノートブックと Python パッケージを含めることを受け入れます。

次に、新しいプロジェクトフォルダーに新しい仮想環境を作成します。

cd my_pydabs_project
uv sync

既定では、テンプレートには、 resources/my_pydabs_project_job.py ファイルに Python として定義されているジョブの例が含まれています。

from databricks.bundles.jobs import Job


my_pydabs_project_job = Job.from_dict(
    {
        "name": "my_pydabs_project_job",
        "tasks": [
            {
                "task_key": "notebook_task",
                "notebook_task": {
                    "notebook_path": "src/notebook.ipynb",
                },
            },
        ],
    },
)

Job.from_dict関数は、YAML と同じ形式を使用して Python ディクショナリを受け入れます。リソースは、データクラス構文を使用して構築することもできます。

from databricks.bundles.jobs import Job, Task, NotebookTask


my_pydabs_project_job = Job(
    name="my_pydabs_project_job",
    tasks=[
        Task(
            task_key="notebook_task",
            notebook_task=NotebookTask(
                notebook_path="src/notebook.ipynb",
            ),
        ),
    ],
)

Python ファイルは、pythonの databricks.yml セクションで指定されたエントリポイントを介して読み込まれます。

python:
  # Activate the virtual environment before loading resources defined in
  # Python. If disabled, it defaults to using the Python interpreter
  # available in the current shell.
  venv_path: .venv
  # Functions called to load resources defined in Python.
  # See resources/__init__.py
  resources:
    - 'resources:load_resources'

既定では、 resources/__init__.py には、リソースパッケージ内のすべての Python ファイルを読み込む関数が含まれています。

from databricks.bundles.core import (
    Bundle,
    Resources,
    load_resources_from_current_package_module,
)


def load_resources(bundle: Bundle) -> Resources:
    """
    'load_resources' function is referenced in databricks.yml and is responsible for loading
    bundle resources defined in Python code. This function is called by Databricks CLI during
    bundle deployment. After deployment, this function is not used.
    """

    # the default implementation loads all Python files in 'resources' directory
    return load_resources_from_current_package_module()

ジョブまたはパイプラインをデプロイして実行する

開発ターゲットにバンドルをデプロイするには、バンドルプロジェクトルートから bundle deploy コマンドを使用します。

databricks bundle deploy --target dev

このコマンドは、バンドルプロジェクトに定義されているすべてのものをデプロイします。たとえば、既定のテンプレートを使用して作成されたプロジェクトは、 [dev yourname] my_pydabs_project_job というジョブをワークスペースにデプロイします。そのジョブを見つけるには、Databricks ワークスペースの [ジョブ] と [パイプライン ] に移動します。

バンドルがデプロイされたら、バンドルの概要コマンドを使用して、デプロイされるすべての内容を確認できます。

databricks bundle summary --target dev

最後に、ジョブまたはパイプラインを実行するには、バンドル実行コマンドを使用します。

databricks bundle run my_pydabs_project_job

既存のバンドルを更新する

既存のバンドルを更新するには、「テンプレートからプロジェクトを作成する」の説明に従ってプロジェクトテンプレート構造をモデル化します。 YAML を含む既存のバンドルは、pythonに databricks.yml セクションを追加することで、Python コードとして定義されたリソースを含むように更新できます。

python:
  # Activate the virtual environment before loading resources defined in
  # Python. If disabled, it defaults to using the Python interpreter
  # available in the current shell.
  venv_path: .venv
  # Functions called to load resources defined in Python.
  # See resources/__init__.py
  resources:
    - 'resources:load_resources'

指定された仮想環境には、インストールされている databricks-bundles PyPi パッケージが含まれている必要があります。

pip install databricks-bundles==0.275.0

resources フォルダーには、 __init__.py ファイルが含まれている必要があります。

from databricks.bundles.core import (
    Bundle,
    Resources,
    load_resources_from_current_package_module,
)


def load_resources(bundle: Bundle) -> Resources:
    """
    'load_resources' function is referenced in databricks.yml and
    is responsible for loading bundle resources defined in Python code.
    This function is called by Databricks CLI during bundle deployment.
    After deployment, this function is not used.
    """

    # default implementation loads all Python files in 'resources' folder
    return load_resources_from_current_package_module()

既存のジョブを Python に変換する

既存のジョブを Python に変換するには、 コードとしての表示 機能を使用できます。「ジョブをコードとして表示する」を参照してください。

Databricks ワークスペース内の既存のジョブのページを開きます。
[ 今すぐ実行 ] ボタンの左側にある kebab をクリックし、[ コードとして表示] をクリックします。
[Python]、[Databricks Asset Bundles] の順に選択します
[ コピー ] をクリックし、生成された Python を Python ファイルとしてバンドルプロジェクトの resources フォルダーに保存します。

ヒント

バンドル構成 YAML ファイルに直接貼り付けることができる既存のジョブとパイプラインの YAML を表示およびコピーすることもできます。

メタデータを使用してリソースを作成する

load_resources関数の既定の実装では、resources パッケージ内の Python ファイルが読み込まれます。 Python を使用して、プログラムによってリソースを作成できます。たとえば、構成ファイルを読み込み、ループでジョブを作成できます。

from databricks.bundles.core import (
    Bundle,
    Resources,
    load_resources_from_current_package_module,
)


from databricks.bundles.jobs import Job


def create_job(country: str):
    my_notebook = {
        "task_key": "my_notebook",
        "notebook_task": {
            "notebook_path": "files/my_notebook.py",
        },
    }


    return Job.from_dict(
        {
            "name": f"my_job_{country}",
            "tasks": [my_notebook],
        }
    )


def load_resources(bundle: Bundle) -> Resources:
    resources = load_resources_from_current_package_module()


    for country in ["US", "NL"]:
        resources.add_resource(f"my_job_{country}", create_job(country))


    return resources

バンドル変数にアクセスする

bundle パラメーターを使用して、バンドル変数とデプロイターゲットにアクセスできます。

from databricks.bundles.core import Bundle, Resources, Variable, variables

@variables
class Variables:
    warehouse_id: Variable[str]


def load_resources(bundle: Bundle) -> Resources:
    warehouse_id = bundle.resolve_variable(Variables.warehouse_id)

    ...

詳細については、「リソースとバンドル」クラスのリファレンスを参照してください。

YAML または Python で定義されているリソースを変更する

リソースを読み込む関数と同様に、 databricks.ymlでミューテーター関数を参照できます。この機能は、Python で定義されているリソースの読み込みとは別に使用でき、YAML と Python の両方で定義されているリソースを変更できます。

まず、次の内容でバンドルルートに mutators.py を作成します。

from dataclasses import replace


from databricks.bundles.core import Bundle, job_mutator
from databricks.bundles.jobs import Job, JobEmailNotifications


@job_mutator
def add_email_notifications(bundle: Bundle, job: Job) -> Job:
    if job.email_notifications:
        return job


    email_notifications = JobEmailNotifications.from_dict(
        {
            "on_failure": ["${workspace.current_user.userName}"],
        }
    )


    return replace(job, email_notifications=email_notifications)

次の構成を使用して、バンドルのデプロイ中に add_email_notifications 関数を実行します。これにより、バンドルで定義されているすべてのジョブが、存在しない場合は電子メール通知で更新されます。ミューテーター関数は databricks.ymlで指定する必要があり、指定された順序で実行されます。ジョブミューテーターは、バンドルで定義されているすべてのジョブに対して実行され、更新されたコピーまたは変更されていない入力を返すことができます。ミューテーターは、既定のジョブクラスターや SQL ウェアハウスの構成など、他のフィールドにも使用できます。

python:
  mutators:
    - 'mutators:add_email_notifications'

ミューテーターの実行中に関数が例外をスローした場合、バンドルのデプロイは中止されます。

フィードバック

このページはお役に立ちましたか?

Last updated on 2025-11-14