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

[Run a Delta Live Tables pipeline in a workflow \| Databricks on AWS](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-workflows.html) [2022/1/12時点]の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

Databricksジョブ、Apache Airflow、Azure Data Factoryのデータ処理ワークフローの一部としてDelta Live Tablesパイプラインを実行することができます。

# ジョブ

データ処理ワークフローを実行するために、Databricksジョブのマルチタスクをオーケストレーションすることができます。ジョブにDelta Live Tablesパイプラインを含めるには、[ジョブを作成](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E3%81%AE%E4%BD%9C%E6%88%90)する際に**Pipeline**タスクを使用します。

# Apache Airflow

[Apache Airflow](https://airflow.apache.org/)は、データワークフローを管理、スケジュールするためのオープンソースのソリューションです。Airflowはワークフローをオペレーションの有向非巡回グラフ(DAG)として表現します。Pythonでワークフローを定義して、Airflowは処理の実行とスケジューリングを管理します、DatabricksでAirflowをインストールし、活用するための情報については[Apache Airflow](https://docs.databricks.com/dev-tools/data-pipelines.html#airflow)をご覧ください。

Airflowワークフローの一部としてDelta Live Tablesパイプラインを実行するためには、[DatabricksSubmitRunOperator](https://airflow.apache.org/docs/apache-airflow-providers-databricks/stable/_api/airflow/providers/databricks/operators/databricks/index.html#airflow.providers.databricks.operators.databricks.DatabricksSubmitRunOperator)を使用してください。

## 要件

AirflowのDelta Live Tablesサポートを使用するには、以下の要件を満足する必要があります。

- Airflowバージョン2.1.0以降
- [Databricks provider](https://pypi.org/project/apache-airflow-providers-databricks/)パッケージバージョン2.1.0以降

## サンプル

以下のサンプルでは、ID`8279d543-063c-4d63-9926-dae38e35ce8b`のDelta Live Tablesパイプラインのアップデートを起動するAirflow DAGを作成しています。

```py: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('dlt',
         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={"pipeline_id": "8279d543-063c-4d63-9926-dae38e35ce8b"}
  )
```

`CONNECTION_ID`をお使いのワークスペースの[Airflowコネクション](https://docs.databricks.com/dev-tools/data-pipelines.html#airflow-connection)のIDで置き換えてください。

このサンプルを`airflow/dags`ディレクトリに保存し、DAGを[参照・実行](https://docs.databricks.com/dev-tools/data-pipelines.html#test-dag)するためにはAirflowのUIを使用します。パイプラインのアップデートの詳細を参照するには、Delta Live TablesのUIを使用します。

# Azure Data Factory

Azure Data Factoryは、データのインテグレーション、変換ワークフローのオーケストレーションを実現するクラウドベースのETLサービスです。Azure Data Factoryは、[ノートブック](https://docs.microsoft.com/azure/data-factory/transform-data-using-databricks-notebook)、JARタスク、Pythonスクリプトを含むワークフローのDatabricksタスクの実行を直接サポートしています。Azure Data Factoryの[Webアクティビティ](https://docs.microsoft.com/azure/data-factory/control-flow-web-activity)からDelta Live Tablesの[API](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-api-guide.html)を呼び出すことで、ワークフローにパイプラインを含めることができます。

1. [データファクトリーを作成](https://docs.microsoft.com/azure/data-factory/quickstart-create-data-factory-portal#create-a-data-factory)するか、既存のデータファクトリーを開きます。
1. 作成が完了したら、データファクトリーのページを開き、**Open Azure Data Factory Studio**タイルをクリックします。Azure Data FactoryのUIが表示されます。
1. Databricks[リンクサービス](https://docs.microsoft.com/azure/data-factory/transform-data-using-databricks-notebook#create-linked-services)を作成します。
1. Azure Data Factory StudioのUIの**New**ドロップダウンから**Pipeline**を選択し、新規Azure Data Factoryパイプラインを作成します。
1. **Activities**ツールボックスで、**General**を展開し、パイプラインキャンバスに**Web**アクティビティをドラッグします。**Settings**タブをクリックし、以下を入力します。
    - **URL**: `https://<databricks-instance>/api/2.0/pipelines/<pipeline-id>/updates` 

    `<databricks-instance>`を[Databricksのワークスペースインスタンス名](https://qiita.com/taka_yayoi/items/7127cabac70dd994fcba#%E3%83%AF%E3%83%BC%E3%82%AF%E3%82%B9%E3%83%9A%E3%83%BC%E3%82%B9%E3%81%AE%E3%82%A4%E3%83%B3%E3%82%B9%E3%82%BF%E3%83%B3%E3%82%B9%E5%90%8Durlid)、例えば、`dbc-a1b2345c-d6e7.cloud.databricks.com`で置き換えてください。

    `<pipeline-id>`をパイプラインIDで置き換えてください。

    - **Method**: ドロップダウンから**POST**を選択します。
    - **Headers**: **+ New**をクリックします。**Name**テキストボックスに**Authorization**を入力します。**Value**に`Bearer <personal-access-token>`を入力します。

    `<personal-access-token>`をDatabricks[パーソナルアクセストークン](https://docs.databricks.com/dev-tools/api/latest/authentication.html)で置き換えてください。

   - **Body**: 追加のリクエストパラメーターを指定するには、パラメーターを含むJSONドキュメントを入力します。例えば、パイプラインの全てのデータのアップデート、再処理を起動するには：`{"full_refresh": "true"}`を指定します。追加のリクエストパラメーターがない場合は、空の中括弧(`{}`)を指定します。

Webアクティビティをテストするには、Data Factory UIのパイプラインツールバーの**Debug**をクリックします。Azure Data Factoryパイプラインの**Output**タブに、エラーを含む出力、実行ステータスが表示されます。パイプラインのアップデートの詳細を参照するにはDelta Live TablesのUIを使用します。

:::note info
**ティップス**
一般的なワークフロー要件には、以前のタスクの完了後にタスクを起動するというものがあります。Delta Live Tablesの`updates`リクエストは非同期であり、リクエストはあっ苦デートが完了した後ではなくアップデートが起動した後に値を返します。Delta Live Tablesのアップデートに依存するお使いのAzure Data Factoryパイプラインは、アップデートが完了するまで待たなくてはなりません。アップデートの完了まで待つためのオプションは、Delta Live Tablesアップデートを起動するWebアクティビティの後に[Untilアクティビティ](https://docs.microsoft.com/azure/data-factory/control-flow-until-activity)を追加するというものです。

1. アップデートの完了後の待ち時間を、[Waitアクティビティ](https://docs.microsoft.com/azure/data-factory/control-flow-wait-activity)に秒数で指定します。
1. Waitアクティビティの後に、アップデートの状態を取得するためにDelta Live Tablesの[Get update details](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-api-guide.html#update-details)リクエストを使用するWebアクティビティを追加します。レスポンスの`state`フィールドには、アップデートの完了を含む、アップデートの状態が含まれます。
1. Untilアクティビティの終了条件を指定するために`state`フィールドの値を使用します。`state`の値に基づいてパイプラインの変数を追加し、この変数を終了条件に使用するために[Set Variableアクティビティ](https://docs.microsoft.com/azure/data-factory/control-flow-set-variable-activity)を使用することもできます。
:::

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)