ワークフローでパイプラインを実行する

Lakeflow Jobs、 Apache Airflow 、またはAzure Data Factory を使用して、データ処理ワークフローの一部としてパイプラインを実行できます。

パイプラインは、データセット間の依存関係を自動的に解決するため、単純なパイプライン内のオーケストレーションを独自に処理します。条件付き実行、タスク結果に基づく分岐、リトライ、他の種類の作業とのパイプラインの調整など、パイプラインが対応していないオーケストレーションの場合、ロジックをパイプラインに組み込むのではなく、専用のワークフローオーケストレーターを使用します。

オーケストレーション用にパイプラインを準備

オーケストレーションは、各パイプラインが個別の作業単位を扱い、それを単独でスケジュールしたり、検証したり、実行したりしたい場合に最適に機能します。これらの境界を考慮してパイプラインを設計することで、ワークフローが、アップストリームタスクとダウンストリームタスク間の適切な制御フローを含め、個別のタスクとして調整できるようになります。

すでに、個別にオーケストレーションしたい作業を組み合わせた大規模なパイプラインがある場合は、テーブルを新しいパイプラインに移動することで、それをより小さなパイプラインに分割してください。「パイプライン間のテーブルの移動」を参照してください。

Lakeflow Jobs

データ処理ワークフローを実装するために、Lakeflow Jobs で複数のタスクをオーケストレーションできます。ジョブにパイプラインを含めるには、ジョブを作成する際に パイプライン タスクを使用します。ジョブのパイプラインタスクを参照してください。

Apache Airflow

Apache Airflow は、データワークフローを管理およびスケジューリングするためのオープンソースソリューションです。Airflow は、ワークフローを操作の有向非巡回グラフ (DAG) として表します。Pythonファイルでワークフローを定義し、Airflowがスケジューリングと実行を管理します。DatabricksでのAirflowのインストールと使用に関する情報については、Apache AirflowによるLakeflowジョブのオーケストレーションを参照してください。

Airflow ワークフローの一部としてパイプラインを実行するには、 DatabricksSubmitRunOperatorを使用します。

要件

LakeFlow PipelinesのAirflowサポートを使用するには、次のものが必要です：

Airflowバージョン2.1.0 またはそれ以降。
Databricks プロバイダーパッケージバージョン 2.1.0またはそれ以降。

例

次の例では、識別子8279d543-063c-4d63-9926-dae38e35ce8bを持つパイプラインの更新をトリガーする Airflow DAG を作成します。

Python
from airflow import DAG
from airflow.providers.databricks.operators.databricks import DatabricksSubmitRunOperator
from airflow.utils.dates import days_ago

default_args = {
  'owner': 'airflow'
}

with DAG('ldp',
         start_date=days_ago(2),
         schedule_interval="@once",
         default_args=default_args
         ) as dag:

  opr_run_now=DatabricksSubmitRunOperator(
    task_id='run_now',
    databricks_conn_id='CONNECTION_ID',
    pipeline_task={&quot;pipeline_id&quot;: &quot;8279d543-063c-4d63-9926-dae38e35ce8b&quot;}
  )

CONNECTION_ID 、ワークスペースへのAirflow 接続の識別子に置き換えます。

この例をairflow/dagsディレクトリに保存し、Airflow UI を使用して DAG を表示およびトリガーします。パイプライン UI を使用して、パイプライン更新の詳細を表示します。

Azure Data Factory

注記

Lakeflow pipelines と Azure Data Factory にはそれぞれ、障害発生時の再試行回数を構成するオプションが含まれています。 パイプライン**と**、パイプラインを呼び出す Azure Data Factory アクティビティで再試行値が構成されている場合、再試行回数は、Azure Data Factory の再試行値にパイプラインの再試行値を掛けた値になります。

たとえば、パイプラインの更新が失敗した場合、パイプラインはdefaultで最大5回更新を再試行します。Azure Data Factoryの再試行が3に設定されており、パイプラインがdefaultの5回再試行を使用する場合、失敗しているパイプラインは最大15回再試行される可能性があります。パイプラインの更新が失敗した際の過度な再試行を回避するため、Databricksは、パイプライン、またはそのパイプラインを呼び出すAzure Data Factoryアクティビティを構成する際に、再試行回数を制限することを推奨します。

パイプラインの再試行設定を変更するには、パイプラインを構成するときにpipelines.numUpdateRetryAttempts設定を使用します。

Azure Data Factory は、データ統合および変換ワークフローを調整できるクラウドベースの ETL サービスです。Azure Data Factory は、ノートブック、JAR タスク、Python スクリプトなどのワークフロー内での Databricks タスクの実行を直接サポートします。Azure Data FactoryWeb アクティビティからパイプライン RESTAPI を呼び出して、ワークフローにパイプラインを含めることもできます。たとえば、Azure Data Factory からパイプラインの更新をトリガーするには、次のようにします。

データファクトリを作成するか、既存のデータファクトリを開きます。
作成が完了したら、データファクトリのページを開き、 Azure Data Factory Studio を開く タイルをクリックします。 Azure データファクトリのユーザーインターフェイスが表示されます。
Azure Data Factory Studio ユーザーインターフェイスの新規ドロップダウンメニューから パイプライン を選択して、新しい Azure Data Factoryパイプラインを作成します。
アクティビティ ツールボックスで、全般を展開し、 Web アクティビティをパイプラインキャンバスにドラッグします。設定タブをクリックし、次の値を入力します。

注記

セキュリティのベストプラクティスとして、自動化されたツール、システム、スクリプト、アプリで認証する場合、Databricks では、ワークスペースユーザーではなく、サービスプリンシパルに属する個人用アクセストークンを使用することをお勧めします。サービスプリンシパルのトークンを作成するには、「サービスプリンシパルのトークンの管理」を参照してください。

URL : https://<databricks-instance>/api/2.0/pipelines/<pipeline-id>/updates 。

<get-workspace-instance>を置き換えます。

<pipeline-id>パイプライン識別子に置き換えます。
方法 : ドロップダウンメニューから POST を選択します。
ヘッダー : + 新規 をクリックします。名前テキストボックスに、 Authorizationと入力します。値テキストボックスに、 Bearer <personal-access-token>と入力します。

<personal-access-token> Databricksの個人アクセス許可に置き換えます。
Body : 追加のリクエストを渡すには、パラメーターを含むJSONドキュメントを入力します。たとえば、更新を開始し、パイプラインのすべてのデータを再処理するには、次のようにします: {"full_refresh": "true"} 。追加のリクエストがない場合は、空の中括弧 ( {} ) を入力してください。

Web アクティビティをテストするには、Data Factory UI のパイプラインツールバーで [デバッグ] をクリックします。エラーを含む実行の出力とステータスは、Azure Data Factory パイプラインの [出力] タブに表示されます。パイプラインの更新の詳細を表示するには、パイプライン UI を使用します。

ヒント

一般的なワークフロー要件は、前のタスクの完了後にタスクを起動することです。パイプラインのupdates要求は非同期であり、更新の開始後に戻りますが、完了前であるため、パイプラインの更新に依存するAzure Data Factoryパイプライン内のタスクは、更新が完了するまで待機する必要があります。更新の完了を待つオプションは、パイプラインの更新をTriggerするWebアクティビティの後にUntilアクティビティを追加することです。Untilアクティビティでは、

更新が完了するまで設定された秒数待機するための待機アクティビティを追加します。
パイプライン更新の詳細要求を使用して更新のステータスを取得する Wait アクティビティの後に Web アクティビティを追加します。レスポンスのstateフィールドには、更新が完了したかどうかも含め、更新の現在の状態が返されます。
stateフィールドの値を使用して、Until アクティビティの終了条件を設定します。変数設定アクティビティを使用して、 state値に基づいてパイプライン変数を追加し、この変数を終了条件に使用することもできます。

オーケストレーション用にパイプラインを準備​

Lakeflow Jobs​

Apache Airflow​

要件​

例​

Azure Data Factory​

オーケストレーション用にパイプラインを準備

Lakeflow Jobs

Apache Airflow

要件

例

Azure Data Factory