Databricks における Databricks Utilities taskValues サブユーティリティの基本的な動作確認

概要

Databricks にてタスク値を使用してタスク間で変数を渡せる Databricks Utilities taskValues サブユーティリティの基本的な動作確認を共有します。

本記事では、taskValues の基本動作を次の 2 つの観点から整理し、動作確認結果を共有します。

1. タスク間での値の受け渡し — dbutils.jobs.taskValues.get で取得する方法と、ジョブパラメーター経由で dbutils.widgets.get により取得する方法の違い
2. For each タスクとの組み合わせ — 動的に生成したリストに対するループ処理の共通化

基本的な情報

タスク値(taskValues)とは

タスク値とは、Databricks Utilities の taskValues サブユーティリティ (dbutils.jobs.taskValues) を指し、Databricks ジョブ内のタスク間で任意の値を受け渡しするための仕組みです。あるタスクで dbutils.jobs.taskValues.set() を使用してキーと値のペアを設定すると、後続のタスクでタスク名とキーを使用してその値を参照できます。

主なコマンドと制約

dbutils.jobs.taskValues サブユーティリティで使用できる主なコマンドは以下の 2 つです。

コマンド	説明
set(key, value)	タスク値を設定または更新する。
get(taskKey, key, default, debugValue)	指定したアップストリーム タスクに設定されたタスク値を取得する。debugValue を指定しておくと、ジョブ実行外（ノートブック単体実行時）でもデフォルト値でエラーなく動作するため、開発時の利便性が高まります。

詳細については下記ドキュメントを参照してください。

• taskValues サブユーティリティ (dbutils.jobs.taskValues) - Azure Databricks | Microsoft Learn

For each タスクとの組み合わせ

タスク値は、 For each タスク と組み合わせることで、動的に生成したリスト(テーブル一覧、顧客 ID リスト、製品 ID リストなど)に対してループ処理を実行するパターンで強力に機能します。アップストリームタスクでリストをタスク値として設定し、For each タスクの [入力] テキストボックスに {{tasks.<task_name>.values.<value_name>}} を指定することで、そのリストを反復処理できます。

主な制約事項

主な制限事項には下記があります。

• 値は JSON で表現可能なもののみ
• タスク値参照で渡す場合: 48 KB

上記の超えるサイズが必要な場合は、値そのものを渡すのではなく、ルックアップテーブルの利用を検討してください。対応方法として、下記の記事にてて解説されています。

• 各タスクにおいて、大規模なパラメーター配列にルックアップテーブルを使用する - Azure Databricks | Microsoft Learn

参考

• タスク値を使用してタスク間で情報を渡す - Azure Databricks | Microsoft Learn
• Databricks Utilities (dbutils) リファレンス - Azure Databricks | Microsoft Learn
• For タスクを使用して別のタスクをループで実行する - Azure Databricks | Microsoft Learn
• 各タスクにおいて、大規模なパラメーター配列にルックアップテーブルを使用する - Azure Databricks | Microsoft Learn

基本的な利用方法

ここでは、1 つの「値を設定するタスク」と 2 つの「値を取得するタスク」を用意し、取得方法の違いによる挙動を比較します。具体的には以下の 3 種類のノートブックを作成します。

ノートブック	役割
set_paras	dbutils.jobs.taskValues.set で値をセットする
get_by_taskValues	dbutils.jobs.taskValues.get で値を直接取得する
get_by_job_parameters	ジョブパラメーター経由で渡された値を dbutils.widgets.get で取得する

事前準備

1. taskValues に値を渡すノートブックを作成

後続タスクに渡したい辞書型の値を set します。末尾の get は、ノートブック単体で実行した際の動作確認用であり、debugValue を指定することでジョブ外でもエラーにならないようにしています。

table_conf = {
    "full_table_name": "samples.tpch.nation"
}

dbutils.jobs.taskValues.set(
    key="table_conf",
    value=table_conf,
)

# 内容確認
dbutils.jobs.taskValues.get(
    taskKey="set_paras",
    key="table_conf",
    debugValue=table_conf,
)

2. dbutils.jobs.taskValues.get により値を取得するノートブックを作成

前のタスクで設定された値を、タスク値として直接取得するパターンです。debugValue は開発時にジョブ外で実行した際のフォールバック値として機能します。

debug_value = {"full_table_name": "test.test.test"}

parameters = dbutils.jobs.taskValues.get(
    taskKey="set_paras",
    key="table_conf",
    debugValue=debug_value,
)
print(parameters)
print(type(parameters))

3. dbutils.widgets.get により値を取得するノートブックを作成

同じ値を、ジョブパラメーター経由で文字列として受け取るパターンです。ジョブ定義側でタスク値参照 ({{tasks.set_paras.values.table_conf}}) をパラメーターにバインドする形になります。

dbutils.widgets.text("table_conf", '{"full_table_name": "test.test.test"}')
table_conf = dbutils.widgets.get("table_conf")

print(table_conf)
print(type(table_conf))

4. ジョブを作成

上記 3 つのノートブックを、set_paras をアップストリームとして下流 2 タスクが並列実行されるように構成します。

resources:
  jobs:
    taskValues:
      name: taskValues
      tasks:
        - task_key: set_paras
          notebook_task:
            notebook_path: /Workspace/qiita/taskValues/set_paras
            source: WORKSPACE
        - task_key: get_by_job_parameters
          depends_on:
            - task_key: set_paras
          notebook_task:
            notebook_path: /Workspace/qiita/taskValues/get_by_job_parameters
            base_parameters:
              table_conf: "{{tasks.set_paras.values.table_conf}}"
            source: WORKSPACE
        - task_key: get_by_taskValues
          depends_on:
            - task_key: set_paras
          notebook_task:
            notebook_path: /Workspace/qiita/taskValues/get_by_taskValues
            source: WORKSPACE
      queue:
        enabled: true
      performance_target: PERFORMANCE_OPTIMIZED

ジョブを実行

ジョブを実行し、3 つのタスクがすべて成功することと、取得側の 2 つのタスクで型の違いが表れることを確認します。

ジョブの実行結果

taskValues に値を渡すノートブックの結果

dbutils.jobs.taskValues.get により値を取得するノートブックの結果

dict 型のまま取得できており、そのままキー参照で値にアクセスできます。

{'full_table_name': 'samples.tpch.nation'}
<class 'dict'>

dbutils.widgets.get により値を取得するノートブックの結果

一方、ジョブパラメーター経由で渡した場合は JSON 文字列 として受け取ることになります。辞書として扱うには json.loads などで明示的にパースする必要がある点に注意してください。

{"full_table_name": "samples.tpch.nation"}
<class 'str'>

taskValues サブユーティリティと For each タスクによる処理の汎用化

実装概要

ここからは、タスク値を使った実用パターンとして、「テーブル一覧を一箇所で定義し、共通処理を各テーブルに対してループ実行する」 構成を実装します。

この構成により、処理対象のテーブルを追加・削除する際は設定用ノートブック (set_table_list) を変更するだけで済み、共通処理側は一切手を入れる必要がありません。

テーブル一覧のイメージ

共通処理のノートブック

テーブル一覧のパラメータを定義したノートブックを作成

For each タスクに渡す入力はリスト形式である必要があります。ここでは辞書のリストとして定義し、各要素を 1 イテレーション分のパラメーターとして扱います。

table_list = [
    {"full_table_name": "samples.tpch.nation"},
    {"full_table_name": "samples.tpch.region"},
    {"full_table_name": "samples.tpch.customer"},
]

dbutils.jobs.taskValues.set(
    key="table_list",
    value=table_list,
)

# 内容確認
dbutils.jobs.taskValues.get(
    taskKey="set_table_list",
    key="table_list",
    debugValue=table_list,
)

共通処理のノートブックを作成

各イテレーションで受け取る full_table_name をウィジェット経由で取得し、該当テーブルを処理します。ここではサンプルとして先頭 10 件を表示するだけの処理にしていますが、実務ではこの部分が取り込み処理や集計処理などに置き換わります。

dbutils.widgets.text("full_table_name", 'samples.nyctaxi.trips')
full_table_name = dbutils.widgets.get("full_table_name")

df = spark.table(full_table_name)
df.limit(10).display()

ジョブを作成

+新規 -> ジョブ を選択します。

テーブル一覧のパラメータを指定したノートブックのタスクを追加します。

+タスクを追加 -> それぞれ（For each）のタスクを選択します。

入力に{{tasks.set_table_list.values.table_list}}という値を入力したタスクを追加します。これがループ対象のリストとなり、table_list の要素数だけイテレーションが発生します。

共通処理のノートブックを指定し、パラメータのキーにfull_table_nameを、値に{{input.full_table_name}}を入力したタスクを追加します。{{input.xxx}} は現在のイテレーションの要素を参照するための予約変数です。

ジョブの定義は下記のようにになります。

resources:
  jobs:
    taskValues_02:
      name: taskValues_02
      tasks:
        - task_key: set_table_list
          notebook_task:
            notebook_path: /Workspace/qiita/taskValues/set_table_list
            source: WORKSPACE
        - task_key: iteration_table_query
          depends_on:
            - task_key: set_table_list
          for_each_task:
            inputs: "{{tasks.set_table_list.values.table_list}}"
            task:
              task_key: query_table
              notebook_task:
                notebook_path: /Workspace/qiita/taskValues/select_query_by_para
                base_parameters:
                  full_table_name: "{{input.full_table_name}}"
                source: WORKSPACE
      queue:
        enabled: true
      performance_target: PERFORMANCE_OPTIMIZED

ジョブを実行

ジョブを実行します。

テーブル一覧のパラメータを指定したノートブックにてtaskValuesの値がセットされていることを確認します。

For eachタスクの結果を確認します。table_list に含まれる 3 テーブル分のイテレーションが実行され、それぞれで共通処理のノートブックが同じロジックで走っていることがわかります。

まとめ

本記事では、Databricks の taskValues サブユーティリティの基本的な動作と、For each タスクと組み合わせた実用的な活用パターンを確認しました。要点は以下のとおりです。

• タスク間で値を受け渡す方法には dbutils.jobs.taskValues.get による直接取得と、ジョブパラメーター経由の dbutils.widgets.get による取得の 2 通りがあり、型の保持とノートブック単体実行のしやすさの観点で使い分ける
• For each タスクとの組み合わせにより、設定と処理を分離した保守性の高いジョブ設計が可能になる
• 大きなパラメーターを扱う場合は 48 KB の上限に注意し、必要に応じてルックアップテーブル方式を検討する

動的なテーブル処理や大量データの並列処理を扱うジョブを設計する際は、ぜひ taskValues と For each タスクの組み合わせを活用してみてください。