Azure Databricks ジョブでタスクを条件付きで実行する

  • [アーティクル]

既定では、ジョブ タスクは依存関係が実行され、すべて成功したときに実行されますが、特定の条件が満たされた場合にのみ実行されるように Azure Databricks ジョブのタスクを構成することもできます。 Azure Databricks ジョブでは、条件付きでタスクを実行するための以下の方法がサポートされています。

  • タスクの依存関係の実行状態に基づいてタスクを実行するための Run if 依存関係を指定できます。 たとえば、Run if を使用して、依存関係の一部またはすべてが失敗した場合でもタスクを実行し、ジョブを失敗から復旧させて実行を継続させることができます。
  • If/else condition タスクは、ブール式の結果に基づいてジョブ DAG の一部を実行するために使用されます。 If/else condition タスクを使用すると、ジョブに分岐ロジックを追加できます。 たとえば、アップストリーム インジェスト タスクが新しいデータを追加する場合にのみ、変換タスクを実行します。 それ以外の場合は、データ処理タスクを実行します。

タスクの Run if 条件を追加する

1 つ以上の依存関係を持つタスクを編集する際に、Run if 条件を設定できます。 タスクにこの条件を追加するには、タスク構成の [Run if 依存関係] ドロップダウン メニューから条件を選択します。 Run if 条件は、タスクのすべての依存関係が完了した後に評価されます。 また、1 つ以上の依存関係を持つ新しいタスクを追加するときに Run if 条件を追加することもできます。

Run if 条件のオプション

次の Run if 条件をタスクに追加できます。

  • すべて成功: すべての依存関係が実行され、成功した。 これは、タスクを実行するためのデフォルトの条件です。 条件が満たされない場合、タスクは Upstream failed としてマークされます。
  • 少なくとも 1 つが成功: 少なくとも 1 つの依存関係が成した。 条件が満たされない場合、タスクは Upstream failed としてマークされます。
  • 失敗なし: 失敗した依存関係がなく、少なくとも 1 つの依存関係が実行された。 条件が満たされない場合、タスクは Upstream failed としてマークされます。
  • すべて完了: タスクは、依存する実行の状態に関係なく、すべての依存関係が実行された後に実行されます。 この条件を使用すると、依存タスクの結果に依存せずに実行されるタスクを定義できます。
  • 少なくとも 1 つが失敗: 少なくとも 1 つの依存関係が失敗した。 条件が満たされない場合、タスクは Excluded としてマークされます。
  • すべて失敗: すべての依存関係が失敗した。 条件が満たされない場合、タスクは Excluded としてマークされます。

Note

  • 失敗を処理するように構成されたタスクは、そのタスクの Run if 条件が満たされない場合、Excluded としてマークされます。 除外されたタスクはスキップされ、成功として扱われます。
  • タスクの依存関係がすべて除外されている場合、Run if 条件に関係なく、そのタスクも除外されます。
  • タスクの実行を取り消すと、その取り消しがダウンストリーム タスクへと伝播し、失敗を処理する Run if 条件を持つタスクが (たとえば、タスクが取り消された際のクリーンアップ タスクの実行を検証するために) 実行されます。

Azure Databricks ジョブがどのようにジョブ実行の状態を判定するか

Azure Databricks ジョブは、ジョブのリーフ タスクの結果に基づいて、ジョブの実行が成功したかどうかを判定します。 リーフ タスクは、ダウンストリームの依存関係を持たないタスクです。 ジョブの実行で、次の 3 つの結果のいずれかが生じます。

  • 成功: すべてのタスクが成功した。
  • 成功したが失敗がある: 一部のタスクが失敗したが、すべてのリーフ タスクは成功した。
  • 失敗: 1 つ以上のリーフ タスクが失敗した。

If/else condition タスクを持つジョブに分岐ロジックを追加する

If/else condition タスクを使用して、ブール式に基づいてジョブ DAG の一部を実行します。 この式はブール演算子とオペランドのペアで構成され、オペランドは、ジョブとタスクのパラメーター変数を使用してジョブまたはタスクの状態を参照したり、タスク値を使用したりする可能性があります。

Note

  • 数値と数値以外の値には、以下のようにブール演算子によって異なる処理が行われます。
    • == および != 演算子は、オペランドの文字列比較を実行します。 たとえば、12.0 == 12 は false に評価されます。
    • >、>=、および <= 演算子はオペランドの数値比較を実行します。 たとえば、12.0 >= 12 は true に評価され、10.0 >= 12 は false に評価されます。
    • オペランドでタスク値を参照する場合は、数値、文字列、およびブール値のみが許可されます。 その他の型では、条件式が失敗します。 数値以外の型は文字列にシリアル化され、If/else condition 式の中では文字列として扱われます。 たとえば、タスク値がブール値に設定されると、タスク値は "true" または "false" にシリアル化されます。

ジョブを作成したり、既存のジョブ内のタスクを編集したりする際に、If/else condition を追加できます。 If/else condition タスクを構成するには:

  1. [種類] ドロップダウン メニューで、If/else condition を選択します。
  2. 最初の [条件] テキスト ボックスに、評価するオペランドを入力します。 オペランドは、ジョブまたはタスクのパラメーター変数またはタスク値を参照できます。
  3. ドロップダウン メニューからブール演算子を選択します。
  4. 2 番目の [条件] テキスト ボックスに、条件を評価するための値を入力します。

If/else condition タスクの依存関係を構成するには:

  1. DAG ビューで If/else condition タスクを選択し、[+ タスクの追加] をクリックします。
  2. タスクの詳細を入力したら、[依存先] をクリックし、<task-name> (true) を選択します。ここで <task-name> は If/else condition タスクの名前です。
  3. false に評価される条件についても同じことを繰り返します。

たとえば、bad_records という名前の値の中の無効なレコードの数を保持する process_records という名前のタスクがあり、無効なレコードが見つかったかどうかに基づいて処理を分岐したいとします。 このロジックをワークフローに追加するには、{{tasks.process_records.values.bad_records}} > 0 のような式を使用して If/else condition タスクを作成します。 その後、条件の結果に基づいて依存タスクを追加できます。

If/else condition タスクを含むジョブの実行が完了すると、UI でジョブ実行の詳細を表示すると、式の結果と式の評価の詳細を表示できます。