Jobs API 2.0 から 2.1 への更新

  • [アーティクル]

Azure Databricks ジョブで複数のタスクをオーケストレーションできるようになりました。 この記事では、複数のタスクを含むジョブをサポートする Jobs API の変更について詳しく説明します。また、この新機能を使用するために既存の API クライアントを更新するのに役立つガイダンスを提供します。

Databricks では、特に複数のタスクを含むジョブを使用する場合に、API スクリプトとクライアント用に Jobs API 2.1 が推奨されます。

この記事では、単一のタスクで定義されたジョブを、''シングルタスク形式'' と呼び、複数のタスクで定義されたジョブを ''マルチタスク形式'' と呼びます。

Jobs API 2.0 および 2.1 では、更新要求がサポートされるようになりました。 シングルタスク形式ジョブとマルチタスク形式ジョブ間の変更を最小限に抑えるために、リセット要求ではなく、update 要求を使用して既存のジョブを変更します。

API の変更

Jobs API では、ジョブ内の各タスクの設定を取り込むために TaskSettings オブジェクトを定義するようになりました。 マルチタスク形式ジョブでは、TaskSettings データ構造の配列である tasks フィールドが、JobSettings オブジェクトに含まれています。 JobSettings のいくつかのフィールドの以前の部分は、マルチタスク形式ジョブのタスク設定に含まれるようになりました。 また、JobSettings は format フィールドを含むように更新されました。 format フィールドはジョブの形式を示しており、STRING 値は SINGLE_TASK または MULTI_TASK に設定されます。

マルチタスク形式ジョブの JobSettings に対するこれらの変更に合わせて、既存の API クライアントを更新する必要があります。 必要な変更の詳細については、API クライアント ガイドを参照してください。

Jobs API 2.1 ではマルチタスク形式がサポートされています。 すべての API 2.1 の要求はマルチタスク形式に準拠する必要があり、応答はマルチタスク形式で構造化されます。 最初に API 2.1 の新機能がリリースされます。

Jobs API 2.0 は、マルチタスク形式ジョブをサポートするために追加フィールドで更新されます。 特に明記されている場合を除き、このドキュメントの例では API 2.0 を使用します。 しかし、Databricks では、新規および既存の API スクリプトとクライアントに対して API 2.1 が推奨されます。

API 2.0 および 2.1 のマルチタスク形式ジョブを表す JSON ドキュメントの例を以下に示します。

{
  "job_id": 53,
  "settings": {
    "name": "A job with multiple tasks",
    "email_notifications": {},
    "timeout_seconds": 0,
    "max_concurrent_runs": 1,
    "tasks": [
      {
        "task_key": "clean_data",
        "description": "Clean and prepare the data",
        "notebook_task": {
          "notebook_path": "/Users/user@databricks.com/clean-data"
        },
        "existing_cluster_id": "1201-my-cluster",
        "max_retries": 3,
        "min_retry_interval_millis": 0,
        "retry_on_timeout": true,
        "timeout_seconds": 3600,
        "email_notifications": {}
      },
      {
        "task_key": "analyze_data",
        "description": "Perform an analysis of the data",
        "notebook_task": {
          "notebook_path": "/Users/user@databricks.com/analyze-data"
        },
        "depends_on": [
          {
            "task_key": "clean_data"
          }
        ],
        "existing_cluster_id": "1201-my-cluster",
        "max_retries": 3,
        "min_retry_interval_millis": 0,
        "retry_on_timeout": true,
        "timeout_seconds": 3600,
        "email_notifications": {}
      }
    ],
    "format": "MULTI_TASK"
  },
  "created_time": 1625841911296,
  "creator_user_name": "user@databricks.com",
  "run_as_user_name": "user@databricks.com"
}

Jobs API 2.1 では、タスク レベルのクラスターあるいは 1 つまたは複数の共有ジョブ クラスターの構成がサポートされています。

  • タスク レベルのクラスターは、タスクの開始時に作成および開始され、タスクの完了時に終了します。
  • 共有ジョブ クラスターを使用すると、同じジョブ内の複数のタスクでクラスターを使用できます。 クラスターは、クラスターを使用する最初のタスクの開始時に作成および開始され、クラスターを使用する最後のタスクの完了後に終了します。 共有ジョブ クラスターは、アイドル時に終了されませんが、それを使用しているすべてのタスクの完了後に終了します。 クラスターを共有する複数の非依存タスクを同時に開始できます。 共有ジョブ クラスターが失敗したか、すべてのタスクが完了する前に終了した場合は、新しいクラスターが作成されます。

共有ジョブ クラスターを構成するには、JobSettings オブジェクトに JobCluster 配列を含めます。 ジョブあたり最大 100 のクラスターを指定できます。 2 つの共有クラスターが構成されているジョブに対する API 2.1 応答の例を以下に示します。

注意

タスクにライブラリの依存関係がある場合は、task フィールドの設定でライブラリを構成する必要があります。共有ジョブ クラスター構成でライブラリを構成することはできません。 次の例では、ingest_orders タスクの構成の libraries フィールドで、ライブラリの依存関係の指定が示されています。

{
  "job_id": 53,
  "settings": {
    "name": "A job with multiple tasks",
    "email_notifications": {},
    "timeout_seconds": 0,
    "max_concurrent_runs": 1,
    "job_clusters": [
      {
        "job_cluster_key": "default_cluster",
        "new_cluster": {
          "spark_version": "7.3.x-scala2.12",
          "node_type_id": "i3.xlarge",
          "spark_conf": {
            "spark.speculation": true
          },
          "aws_attributes": {
            "availability": "SPOT",
            "zone_id": "us-west-2a"
          },
          "autoscale": {
            "min_workers": 2,
            "max_workers": 8
          }
        }
      },
      {
        "job_cluster_key": "data_processing_cluster",
        "new_cluster": {
          "spark_version": "7.3.x-scala2.12",
          "node_type_id": "r4.2xlarge",
          "spark_conf": {
            "spark.speculation": true
          },
          "aws_attributes": {
            "availability": "SPOT",
            "zone_id": "us-west-2a"
          },
          "autoscale": {
            "min_workers": 8,
            "max_workers": 16
          }
        }
      }
    ],
    "tasks": [
      {
        "task_key": "ingest_orders",
        "description": "Ingest order data",
        "depends_on": [ ],
        "job_cluster_key": "auto_scaling_cluster",
        "spark_jar_task": {
          "main_class_name": "com.databricks.OrdersIngest",
          "parameters": [
            "--data",
            "dbfs:/path/to/order-data.json"
          ]
        },
        "libraries": [
          {
            "jar": "dbfs:/mnt/databricks/OrderIngest.jar"
          }
        ],
        "timeout_seconds": 86400,
        "max_retries": 3,
        "min_retry_interval_millis": 2000,
        "retry_on_timeout": false
      },
      {
        "task_key": "clean_orders",
        "description": "Clean and prepare the order data",
        "notebook_task": {
          "notebook_path": "/Users/user@databricks.com/clean-data"
        },
        "job_cluster_key": "default_cluster",
        "max_retries": 3,
        "min_retry_interval_millis": 0,
        "retry_on_timeout": true,
        "timeout_seconds": 3600,
        "email_notifications": {}
      },
      {
        "task_key": "analyze_orders",
        "description": "Perform an analysis of the order data",
        "notebook_task": {
          "notebook_path": "/Users/user@databricks.com/analyze-data"
        },
        "depends_on": [
          {
            "task_key": "clean_data"
          }
        ],
        "job_cluster_key": "data_processing_cluster",
        "max_retries": 3,
        "min_retry_interval_millis": 0,
        "retry_on_timeout": true,
        "timeout_seconds": 3600,
        "email_notifications": {}
      }
    ],
    "format": "MULTI_TASK"
  },
  "created_time": 1625841911296,
  "creator_user_name": "user@databricks.com",
  "run_as_user_name": "user@databricks.com"
}

シングルタスク形式ジョブでは、format フィールドが追加されることを除き、JobSettings データ構造は変わらないままです。 TaskSettings 配列は含まれず、タスクの設定は、JobSettings データ構造の最上位で定義されたままです。 シングルタスク形式ジョブを処理するために、既存の API クライアントに変更を加える必要はありません。

API 2.0 のシングルタスク形式ジョブを表す JSON ドキュメントの例を以下に示します。

{
  "job_id": 27,
  "settings": {
    "name": "Example notebook",
    "existing_cluster_id": "1201-my-cluster",
    "libraries": [
      {
        "jar": "dbfs:/FileStore/jars/spark_examples.jar"
      }
    ],
    "email_notifications": {},
    "timeout_seconds": 0,
    "schedule": {
      "quartz_cron_expression": "0 0 0 * * ?",
      "timezone_id": "US/Pacific",
      "pause_status": "UNPAUSED"
    },
    "notebook_task": {
      "notebook_path": "/notebooks/example-notebook",
      "revision_timestamp": 0
    },
    "max_concurrent_runs": 1,
    "format": "SINGLE_TASK"
  },
  "created_time": 1504128821443,
  "creator_user_name": "user@databricks.com"
}

API クライアント ガイド

このセクションでは、新しいマルチタスク形式機能の影響を受ける API 呼び出しに関するガイドライン、例、および必要な変更について説明します。

このセクションの内容は次のとおりです。

作成

Jobs API で 新しいジョブの作成操作 (POST /jobs/create) を使用してシングルタスク形式ジョブを作成するために、既存のクライアントを変更する必要はありません。

マルチタスク形式ジョブを作成するには、JobSettings の tasks フィールドを使用して各タスクの設定を指定します。 次の例では、2 つのノートブック タスクを含むジョブを作成します。 この例は、API 2.0 および 2.1 用です。

注意

ジョブごとに最大 100 のタスクを指定できます。

{
  "name": "Multi-task-job",
  "max_concurrent_runs": 1,
  "tasks": [
    {
      "task_key": "clean_data",
      "description": "Clean and prepare the data",
      "notebook_task": {
        "notebook_path": "/Users/user@databricks.com/clean-data"
      },
      "existing_cluster_id": "1201-my-cluster",
      "timeout_seconds": 3600,
      "max_retries": 3,
      "retry_on_timeout": true
    },
    {
      "task_key": "analyze_data",
      "description": "Perform an analysis of the data",
      "notebook_task": {
        "notebook_path": "/Users/user@databricks.com/analyze-data"
      },
      "depends_on": [
        {
          "task_key": "clean_data"
        }
      ],
      "existing_cluster_id": "1201-my-cluster",
      "timeout_seconds": 3600,
      "max_retries": 3,
      "retry_on_timeout": true
    }
  ]
}

実行の送信

Jobs API で 1 回限りの実行の作成とトリガー操作 (POST /runs/submit) を使用して、シングルタスク形式ジョブの 1 回限りの実行を送信するために、既存のクライアントを変更する必要はありません。

マルチタスク形式ジョブの 1 回限りの実行を送信するには、JobSettings の tasks フィールドを使用して、クラスターを含む各タスクの設定を指定します。 マルチタスク形式ジョブを送信するときには、タスク レベルでクラスターを設定する必要があります。これは、runs submit 要求で共有ジョブ クラスターがサポートされていないためです。 複数のタスクを指定する JobSettings の例については、「作成」を参照してください。

更新

Jobs API でジョブを部分的に更新する操作 (POST /jobs/update) を使用して、シングルタスク形式ジョブを更新するために、既存のクライアントを変更する必要はありません。

マルチタスク形式ジョブの設定を更新するには、一意の task_key フィールドを使用して新しい task 設定を識別する必要があります。 複数のタスクを指定する JobSettings の例については、「作成」を参照してください。

リセット

Jobs API で ジョブのすべての設定を上書きする操作 (POST /jobs/reset) を使用して、シングルタスク形式ジョブの設定を上書きするために、既存のクライアントを変更する必要はありません。

マルチタスク形式ジョブの設定を上書きするには、TaskSettings データ構造の配列を使用して JobSettings データ構造を指定します。 複数のタスクを指定する JobSettings の例については、「作成」を参照してください。

シングルタスクからマルチタスク形式に切り替えずに個々のフィールドを変更するには、更新を使用します。

リスト

シングルタスク形式ジョブでは、Jobs API の すべてのジョブの一覧表示操作 (GET /jobs/list) からの応答を処理するために、クライアントの変更は必要ありません。

マルチタスク形式ジョブの場合、ほとんどの設定は、ジョブ レベルではなく、タスク レベルで定義されます。 クラスター構成は、タスク レベルまたはジョブ レベルで設定できます。 Job 構造で返されるマルチタスク形式ジョブのクラスターまたはタスク設定にアクセスするようにクライアントを変更するには、次のようにします。

  • マルチタスク形式ジョブの job_id フィールドを解析します。
  • Jobs API でジョブの取得操作 (GET /jobs/get) に job_id を渡して、ジョブの詳細を取得します。 マルチタスク形式ジョブの Get API 呼び出しからの応答例については、「取得」を参照してください。

次の例は、シングルタスクおよびマルチタスク形式ジョブを含む応答を示しています。 この例は API 2.0 用です。

{
  "jobs": [
    {
      "job_id": 36,
      "settings": {
        "name": "A job with a single task",
        "existing_cluster_id": "1201-my-cluster",
        "email_notifications": {},
        "timeout_seconds": 0,
        "notebook_task": {
          "notebook_path": "/Users/user@databricks.com/example-notebook",
          "revision_timestamp": 0
        },
        "max_concurrent_runs": 1,
        "format": "SINGLE_TASK"
      },
      "created_time": 1505427148390,
      "creator_user_name": "user@databricks.com"
    },
    {
      "job_id": 53,
      "settings": {
        "name": "A job with multiple tasks",
        "email_notifications": {},
        "timeout_seconds": 0,
        "max_concurrent_runs": 1,
        "format": "MULTI_TASK"
      },
      "created_time": 1625841911296,
      "creator_user_name": "user@databricks.com"
    }
  ]
}

Get

シングルタスク形式ジョブでは、Jobs API のジョブの取得操作 (GET /jobs/get) からの応答を処理するために、クライアントの変更は必要ありません。

マルチタスク形式ジョブからは、タスク設定を含む task データ構造の配列が返されます。 タスク レベルの詳細にアクセスする必要がある場合は、tasks 配列を反復処理し、必要なフィールドを抽出するようにクライアントを変更する必要があります。

マルチタスク形式ジョブの Get API 呼び出しからの応答の例を以下に示します。 この例は、API 2.0 および 2.1 用です。

{
  "job_id": 53,
  "settings": {
    "name": "A job with multiple tasks",
    "email_notifications": {},
    "timeout_seconds": 0,
    "max_concurrent_runs": 1,
    "tasks": [
      {
        "task_key": "clean_data",
        "description": "Clean and prepare the data",
        "notebook_task": {
          "notebook_path": "/Users/user@databricks.com/clean-data"
        },
        "existing_cluster_id": "1201-my-cluster",
        "max_retries": 3,
        "min_retry_interval_millis": 0,
        "retry_on_timeout": true,
        "timeout_seconds": 3600,
        "email_notifications": {}
      },
      {
        "task_key": "analyze_data",
        "description": "Perform an analysis of the data",
        "notebook_task": {
          "notebook_path": "/Users/user@databricks.com/analyze-data"
        },
        "depends_on": [
          {
            "task_key": "clean_data"
          }
        ],
        "existing_cluster_id": "1201-my-cluster",
        "max_retries": 3,
        "min_retry_interval_millis": 0,
        "retry_on_timeout": true,
        "timeout_seconds": 3600,
        "email_notifications": {}
      }
    ],
    "format": "MULTI_TASK"
  },
  "created_time": 1625841911296,
  "creator_user_name": "user@databricks.com",
  "run_as_user_name": "user@databricks.com"
}

実行の取得

シングルタスク形式ジョブでは、Jobs API の ジョブの実行の取得操作 (GET /jobs/runs/get) からの応答を処理するために、クライアントの変更は必要ありません。

マルチタスク形式ジョブの実行の応答には、TaskSettings の配列が含まれています。 各タスクの実行結果を取得するには、次のようにします。

  • 各タスクを反復処理します。
  • 各タスクの run_id を解析します。
  • run_id を使用して実行の出力の取得操作 (GET /jobs/runs/get-output) を呼び出して、各タスクの実行の詳細を取得します。 この要求からの応答の例を以下に示します。
{
  "job_id": 53,
  "run_id": 759600,
  "number_in_job": 7,
  "original_attempt_run_id": 759600,
  "state": {
    "life_cycle_state": "TERMINATED",
    "result_state": "SUCCESS",
    "state_message": ""
  },
  "cluster_spec": {},
  "start_time": 1595943854860,
  "setup_duration": 0,
  "execution_duration": 0,
  "cleanup_duration": 0,
  "trigger": "ONE_TIME",
  "creator_user_name": "user@databricks.com",
  "run_name": "Query logs",
  "run_type": "JOB_RUN",
  "tasks": [
    {
      "run_id": 759601,
      "task_key": "query-logs",
      "description": "Query session logs",
      "notebook_task": {
        "notebook_path": "/Users/user@databricks.com/log-query"
      },
      "existing_cluster_id": "1201-my-cluster",
      "state": {
        "life_cycle_state": "TERMINATED",
        "result_state": "SUCCESS",
        "state_message": ""
      }
    },
    {
      "run_id": 759602,
      "task_key": "validate_output",
      "description": "Validate query output",
      "depends_on": [
        {
          "task_key": "query-logs"
        }
      ],
      "notebook_task": {
        "notebook_path": "/Users/user@databricks.com/validate-query-results"
      },
      "existing_cluster_id": "1201-my-cluster",
      "state": {
        "life_cycle_state": "TERMINATED",
        "result_state": "SUCCESS",
        "state_message": ""
      }
    }
  ],
  "format": "MULTI_TASK"
}

実行による出力の取得

シングルタスク形式ジョブでは、Jobs API の実行の出力の取得操作 (GET /jobs/runs/get-output) からの応答を処理するために、クライアントの変更は必要ありません。

マルチタスク形式ジョブの場合、親の実行で Runs get output を呼び出すとエラーが発生します。これは、実行の出力が個々のタスクに対してのみ可能であるためです。 マルチタスク形式ジョブの出力とメタデータを取得するには、次のようにします。

  • 実行の出力の取得要求を呼び出します。
  • 応答内の子 run_id フィールドを反復処理します。
  • 子 run_id の値を使用して Runs get output を呼び出します。

実行リスト

シングルタスク形式ジョブでは、ジョブの実行の一覧表示操作 (GET /jobs/runs/list) からの応答を処理するために、クライアントの変更は必要ありません。

マルチタスク形式ジョブの場合は、空の tasks 配列が返されます。 タスクを取得するには、run_id をジョブの実行の取得操作 (GET /jobs/runs/get) に渡します。 マルチタスク形式ジョブの Runs list API 呼び出しからの応答の例を以下に示します。

{
  "runs": [
    {
      "job_id": 53,
      "run_id": 759600,
      "number_in_job": 7,
      "original_attempt_run_id": 759600,
      "state": {
          "life_cycle_state": "TERMINATED",
          "result_state": "SUCCESS",
          "state_message": ""
      },
      "cluster_spec": {},
      "start_time": 1595943854860,
      "setup_duration": 0,
      "execution_duration": 0,
      "cleanup_duration": 0,
      "trigger": "ONE_TIME",
      "creator_user_name": "user@databricks.com",
      "run_name": "Query logs",
      "run_type": "JOB_RUN",
      "tasks": [],
      "format": "MULTI_TASK"
    }
  ],
  "has_more": false
}