Databricks アセット バンドルのジョブにタスクを追加する

このページでは、Databricks アセット バンドルでジョブ タスクを定義する方法について説明します。 ジョブ タスクの詳細については、「 Lakeflow ジョブのタスクの構成と編集」を参照してください。

タスクの構成

ジョブ定義の tasks キーで、バンドル内のジョブのタスクを定義します。 使用可能なタスクの種類のタスク構成の例は、[ タスクの設定 ] セクションにあります。 バンドル内のジョブの定義については、 ジョブを参照してください。

タスク値を設定するために、ほとんどのジョブ タスクの種類にはタスク固有のパラメーターがありますが、タスクに渡される ジョブ パラメーター を定義することもできます。 動的値参照 は、ジョブ パラメーターでサポートされており、タスク間でのジョブ実行に固有の値を渡すことができます。 タスクの種類別にタスク値を渡す方法の詳細については、「タスクの種類 別の詳細」を参照してください。

また、一般的なジョブ タスクの設定をターゲット ワークスペースの設定でオーバーライドすることもできます。 ターゲット設定でのオーバーライドを参照してください。

次の構成例では、2 つのノートブック タスクを含むジョブを定義し、最初のタスクから 2 番目のタスクにタスク値を渡します。

resources:
  jobs:
    pass_task_values_job:
      name: pass_task_values_job
      tasks:
        # Output task
        - task_key: output_value
          notebook_task:
            notebook_path: ../src/output_notebook.ipynb

        # Input task
        - task_key: input_value
          depends_on:
            - task_key: output_value
          notebook_task:
            notebook_path: ../src/input_notebook.ipynb
            base_parameters:
              received_message: '{{tasks.output_value.values.message}}'

output_notebook.ipynbには、message キーのタスク値を設定する次のコードが含まれています。

# Databricks notebook source
# This first task sets a simple output value.

message = "Hello from the first task"

# Set the message to be used by other tasks
dbutils.jobs.taskValues.set(key="message", value=message)

print(f"Produced message: {message}")

input_notebook.ipynbは、タスクの構成で設定されたパラメーター received_messageの値を取得します。

# This notebook receives the message as a parameter.

dbutils.widgets.text("received_message", "")
received_message = dbutils.widgets.get("received_message")

print(f"Received message: {received_message}")

タスクの設定

このセクションには、各ジョブ タスクの種類の設定と例が含まれています。

クリーン ルーム ノートブック タスク

clean_rooms_notebook_task フィールドが存在する場合、クリーン ルーム ノートブック タスクはクリーン ルーム ノートブックを実行します。 クリーン ルームの詳細については、「 Azure Databricks Clean Rooms とは」を参照してください。

クリーン ルーム ノートブック タスクでは、次のキーを使用できます。 対応する REST API オブジェクト定義については、 clean_rooms_notebook_taskを参照してください。

条件タスク

condition_task を使用すると、if/else 条件付きロジックを持つタスクをジョブに追加できます。 タスクは、他のタスクの実行を制御するために使用できる条件を評価します。 条件タスクは、クラスターの実行を必要とせず、再試行や通知をサポートしません。 if/else 条件タスクの詳細については、「 If/else タスクを使用してジョブに分岐ロジックを追加する」を参照してください。

条件タスクには、次のキーを使用できます。 対応する REST API オブジェクト定義については、 condition_taskを参照してください。

例示

次の例には、条件タスクとノートブック タスクが含まれています。ノートブック タスクは、ジョブの修復の数が 5 未満の場合にのみ実行されます。

resources:
  jobs:
    my-job:
      name: my-job
      tasks:
        - task_key: condition_task
          condition_task:
            op: LESS_THAN
            left: '{{job.repair_count}}'
            right: '5'
        - task_key: notebook_task
          depends_on:
            - task_key: condition_task
              outcome: 'true'
          notebook_task:
            notebook_path: ../src/notebook.ipynb

ダッシュボード タスク

このタスクを使用して、ダッシュボードを更新し、スナップショットをサブスクライバーに送信します。 バンドル内のダッシュボードの詳細については、 ダッシュボードを参照してください。

ダッシュボード タスクでは、次のキーを使用できます。 対応する REST API オブジェクト定義については、 dashboard_taskを参照してください。

例示

次の例では、ダッシュボード タスクをジョブに追加します。 ジョブが実行されると、指定した ID のダッシュボードが更新されます。

resources:
  jobs:
    my-dashboard-job:
      name: my-dashboard-job
      tasks:
        - task_key: my-dashboard-task
          dashboard_task:
            dashboard_id: 11111111-1111-1111-1111-111111111111

dbt タスク

このタスクを使用して、1 つ以上の dbt コマンドを実行します。 dbt の詳細については、「 dbt Cloud への接続」を参照してください。

dbt タスクでは、次のキーを使用できます。 対応する REST API オブジェクト定義については、 dbt_taskを参照してください。

例示

次の例では、dbt タスクをジョブに追加します。 この dbt タスクは、指定した SQL ウェアハウスを使用して、指定した dbt コマンドを実行します。

SQL ウェアハウスの ID を取得するには、SQL ウェアハウスの設定ページを開き、[概要] タブの [名前] フィールドで、倉庫の名前の後にかっこで囲まれた ID をコピーします。

resources:
  jobs:
    my-dbt-job:
      name: my-dbt-job
      tasks:
        - task_key: my-dbt-task
          dbt_task:
            commands:
              - 'dbt deps'
              - 'dbt seed'
              - 'dbt run'
            project_directory: /Users/someone@example.com/Testing
            warehouse_id: 1a111111a1111aa1
          libraries:
            - pypi:
                package: 'dbt-databricks>=1.0.0,<2.0.0'

タスクごとに

for_each_task を使用すると、for each ループを使用したタスクをジョブに追加できます。 このタスクは、指定されたすべての入力に対して入れ子になったタスクを実行します。 for_each_taskの詳細については、「For each タスクを使用して別のタスクをループで実行する」を参照してください。

for_each_taskでは、次のキーを使用できます。 対応する REST API オブジェクト定義については、 for_each_taskを参照してください。

例示

次の例では、ジョブに for_each_task を追加します。ジョブは、別のタスクの値を順番にループし、それらを処理します。

resources:
  jobs:
    my_job:
      name: my_job
      tasks:
        - task_key: generate_countries_list
          notebook_task:
            notebook_path: ../src/generate_countries_list.ipnyb
        - task_key: process_countries
          depends_on:
            - task_key: generate_countries_list
          for_each_task:
            inputs: '{{tasks.generate_countries_list.values.countries}}'
            task:
              task_key: process_countries_iteration
              notebook_task:
                notebook_path: ../src/process_countries_notebook.ipnyb

JAR タスク

このタスクを使用して JAR を実行します。 ローカル JAR ライブラリ、またはワークスペース内のライブラリ、Unity カタログ ボリューム、または外部クラウド ストレージの場所を参照できます。 JAR ファイル (Java または Scala) を参照してください。

標準アクセス モードで Unity カタログ対応クラスターに Scala JAR ファイルをコンパイルしてデプロイする方法の詳細については、「 チュートリアル: サーバーレス コンピューティングで Scala コードを実行する」を参照してください。

JAR タスクでは、次のキーを使用できます。 対応する REST API オブジェクト定義については、 jar_taskを参照してください。

例示

次の例では、JAR タスクをジョブに追加します。 JAR のパスはボリュームの位置にあります。

resources:
  jobs:
    my-jar-job:
      name: my-jar-job
      tasks:
        - task_key: my-jar-task
          spark_jar_task:
            main_class_name: org.example.com.Main
          libraries:
            - jar: /Volumes/main/default/my-volume/my-project-0.1.0-SNAPSHOT.jar

ノートブック タスク

このタスクを使用してノートブックを実行します。 「ジョブのノートブック タスク」を参照してください。

ノートブック タスクでは、次のキーを使用できます。 対応する REST API オブジェクト定義については、 notebook_taskを参照してください。

例示

次の例では、ノートブック タスクをジョブに追加し、my_job_run_id という名前のジョブ パラメーターを設定します。 デプロイするノートブックのパスは、このタスクが宣言されている構成ファイルに対する相対パスです。 このタスクは、Azure Databricks ワークスペース内のデプロイされた場所からノートブックを取得します。

resources:
  jobs:
    my-notebook-job:
      name: my-notebook-job
      tasks:
        - task_key: my-notebook-task
          notebook_task:
            notebook_path: ./my-notebook.ipynb
      parameters:
        - name: my_job_run_id
          default: '{{job.run_id}}'

パイプラインのタスク

このタスクを使用してパイプラインを実行します。 「Lakeflow Spark 宣言型パイプライン」を参照してください。

パイプライン タスクには、次のキーを使用できます。 対応する REST API オブジェクト定義については、 pipeline_taskを参照してください。

例示

次の例では、パイプライン タスクをジョブに追加します。 このタスクは、指定されたパイプラインを実行します。

resources:
  jobs:
    my-pipeline-job:
      name: my-pipeline-job
      tasks:
        - task_key: my-pipeline-task
          pipeline_task:
            pipeline_id: 11111111-1111-1111-1111-111111111111

Power BI タスク

このタスクを使用して、Power BI セマンティック モデル (旧称データセット) の更新をトリガーします。

Power BI タスクでは、次のキーを使用できます。 対応する REST API オブジェクト定義については、 power_bi_taskを参照してください。

例示

次の例では、接続、更新する Power BI モデル、エクスポートする Databricks テーブルを指定する Power BI タスクを定義します。

resources:
  jobs:
    my_job:
      name: my_job
      tasks:
        - task_key: power_bi_task
          power_bi_task:
            connection_resource_name: 'connection_name'
            power_bi_model:
              workspace_name: 'workspace_name'
              model_name: 'model_name'
              storage_mode: 'DIRECT_QUERY'
              authentication_method: 'OAUTH'
              overwrite_existing: false
            refresh_after_update: false
            tables:
              - catalog: 'main'
                schema: 'tpch'
                name: 'customers'
                storage_mode: 'DIRECT_QUERY'
            warehouse_id: '1a111111a1111aa1'

Python スクリプト タスク

このタスクを使用して、Python ファイルを実行します。

Python スクリプト タスクでは、次のキーを使用できます。 対応する REST API オブジェクト定義については、 python_taskを参照してください。

例示

次の例では、Python スクリプト タスクをジョブに追加します。 デプロイする Python ファイルのパスは、このタスクが宣言されている構成ファイルに対する相対パスです。 このタスクは、Azure Databricks ワークスペース内のデプロイされた場所から Python ファイルを取得します。

resources:
  jobs:
    my-python-script-job:
      name: my-python-script-job

      tasks:
        - task_key: my-python-script-task
          spark_python_task:
            python_file: ./my-script.py

Python ホイール タスク

このタスクを使用して、Python ホイールを実行します。 Databricks アセット バンドルを使用した Python ホイール ファイルの作成を参照してください。

Python ホイール タスクでは、次のキーを使用できます。 対応する REST API オブジェクト定義については、 python_wheel_taskを参照してください。

例示

次の例では、Python ホイール タスクをジョブに追加します。 デプロイする Python ホイール ファイルのパスは、このタスクが宣言されている構成ファイルに対する相対パスです。 Databricks アセット バンドル ライブラリの依存関係を参照してください。

resources:
  jobs:
    my-python-wheel-job:
      name: my-python-wheel-job
      tasks:
        - task_key: my-python-wheel-task
          python_wheel_task:
            entry_point: run
            package_name: my_package
          libraries:
            - whl: ./my_package/dist/my_package-*.whl

ジョブタスクを実行する

このタスクを使って、別のジョブを実行します。

実行ジョブ タスクには、次のキーを使用できます。 対応する REST API オブジェクト定義については、 run_job_taskを参照してください。

例示

次の例では、最初のジョブを実行する実行ジョブ タスクが、2 つめのジョブに含まれています。

この例では、 置換 を使用して、実行するジョブの ID を取得します。 UI からジョブの ID を取得するには、ワークスペースでジョブを開き、ジョブの設定ページの [ジョブ の詳細] タブの [ジョブ ID] の値から ID をコピーします。

resources:
  jobs:
    my-first-job:
      name: my-first-job
      tasks:
        - task_key: my-first-job-task
          new_cluster:
            spark_version: '13.3.x-scala2.12'
            node_type_id: 'i3.xlarge'
            num_workers: 2
          notebook_task:
            notebook_path: ./src/test.py
    my_second_job:
      name: my-second-job
      tasks:
        - task_key: my-second-job-task
          run_job_task:
            job_id: ${resources.jobs.my-first-job.id}

SQL タスク

このタスクを使用して、SQL ファイル、クエリ、またはアラートを実行します。

SQL タスクでは、次のキーを使用できます。 対応する REST API オブジェクト定義については、 sql_taskを参照してください。

例示

次の例では、SQL ファイル タスクをジョブに追加します。 この SQL ファイル タスクは、指定した SQL ウェアハウスを使用して、指定した SQL ファイルを実行します。

resources:
  jobs:
    my-sql-file-job:
      name: my-sql-file-job
      tasks:
        - task_key: my-sql-file-task
          sql_task:
            file:
              path: /Users/someone@example.com/hello-world.sql
              source: WORKSPACE
            warehouse_id: 1a111111a1111aa1

次の例では、SQL アラート タスクをジョブに追加します。 この SQL アラート タスクでは、指定した SQL ウェアハウスを使用して、指定した SQL アラートを更新します。

resources:
  jobs:
    my-sql-file-job:
      name: my-sql-alert-job
      tasks:
        - task_key: my-sql-alert-task
          sql_task:
            warehouse_id: 1a111111a1111aa1
            alert:
              alert_id: 11111111-1111-1111-1111-111111111111

次の例では、SQL クエリ タスクをジョブに追加します。 この SQL クエリ タスクは、指定した SQL ウェアハウスを使用して、指定された SQL クエリを実行します。

resources:
  jobs:
    my-sql-query-job:
      name: my-sql-query-job
      tasks:
        - task_key: my-sql-query-task
          sql_task:
            warehouse_id: 1a111111a1111aa1
            query:
              query_id: 11111111-1111-1111-1111-111111111111

その他のタスク設定

次のタスク設定を使用すると、すべてのタスクの動作を構成できます。 対応する REST API オブジェクト定義については、 タスクを参照してください。