Delta Live Tablesパイプラインの作成、実行、管理

[Create, run, and manage Delta Live Tables pipelines \| Databricks on AWS](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-ui.html) [2022/8/1時点]の翻訳です。

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

UIやDelta Live Tables APIを用いて、Delta Live Tablesパイプラインの作成、実行、管理、モニタリングをすることができます。また、Databricks[ジョブ](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a)のようなオーケストレーションツールを用いてパイプラインを実行することができます。本書では、UIを用いてDelta Live Tablesのタスクを実行することにフォーカスしています。APIの利用に関しては[API guide](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-api-guide.html)をご覧ください。

最初のパイプラインを作成し実行するには、[Delta Live Tablesクイックスタート ](https://qiita.com/taka_yayoi/items/7fe8ed2c2f95fd53cc3d)をご覧ください。

# パイプラインの作成

1. 以下のいずれかを行います。
    - サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png)**Workflows**をクリックし、**Delta Live Tables**タブをクリックし、![](https://docs.databricks.com/_images/dlt-create-pipeline-button.png)をクリックします。**Create Pipeline**ダイアログが表示されます。
    - サイドバーで![](https://docs.databricks.com/_images/create-icon.png)**Create**をクリックし、メニューから**Pipeline**を選択します。
1. **Product Edition**ドロップダウンからDelta Live Tablesの製品エディションを選択します。
    製品エディションを用いることで、お使いのパイプラインの要件に基づいて最適な製品オプションを選択することができます。[製品エディション](https://qiita.com/taka_yayoi/items/80ee29197d26e177bbeb#%E8%A3%BD%E5%93%81%E3%82%A8%E3%83%87%E3%82%A3%E3%82%B7%E3%83%A7%E3%83%B3)をご覧ください。
1. **Pipeline Name**フィールドにパイプラインの名称を入力します。
1. **Notebook Libraries**フィールドにパイプラインのクエリーを含むノートブックのパスを入力するか、ノートブックをブラウズするために![](https://docs.databricks.com/_images/file-picker.png)をクリックします。
1. パイプラインに追加のノートブックを指定するには、**Add notebook library**ボタンをクリックします。

    ノートブックは任意の順序で追加できます。Delta Live Tablesはパイプラインの処理グラフ構造を構成するために、自動でデータセットの依存関係を解析します。

1. パイプラインで実行されるクラスターにSpark設定を追加するには、**Add configuration**ボタンをクリックします。
1. 検索、クエリーを行えるようにオプションとして、**Target**フィールドにデータベース名を指定することができます。[データセットの公開](#データセットの公開)をご覧下さい。
1. オプションとして、**Storage Location**フィールドにパイプラインから出力されるデータの格納場所をDBFSからクラウドストレージパスを指定します。**Storage Location**が空の場合、システムはデフォルトの格納場所を使用します。
1. **Pipeline Mode**に**Triggered**か**Continuous**を選択します。[連続、トリガーパイプライン](https://qiita.com/taka_yayoi/items/80ee29197d26e177bbeb#%E9%80%A3%E7%B6%9A%E3%83%88%E3%83%AA%E3%82%AC%E3%83%BC%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3)をご覧下さい。
1. オプションとして、オートスケーリングの有効・無効、ワーカーノード数の設定などパイプラインのクラスターの設定を変更することができます。[クラスターサイズの管理](#クラスターサイズの管理)をご覧ください。
1. オプションとして、パイプラインで[Photonランタイム](https://qiita.com/taka_yayoi/items/6fdbcac84f88bdd14fee)を使用することができます。**Use Photon Acceleration**のチェックボックスをクリックしてください。
1. オプションとして、このパイプラインのDelta Live Tablesランタイムバージョンを変更できます。**Channel**ドロップダウンをクリックしてください。Delta Live Tablesの[settings](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-configuration.html#config-settings)の`channel`フィールドをご覧くださいさい。
1. **Create**をクリックします。

オプションとして、パイプラインのJSON設定を参照、編集することができます。**Create Pipeline**ダイアログの**JSON**をクリックします。

# パイプラインのアップデートのスタート

作成したパイプラインを実行するには、[パイプラインのアップデート](https://qiita.com/taka_yayoi/items/80ee29197d26e177bbeb#%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%81%AE%E3%82%A2%E3%83%83%E3%83%97%E3%83%87%E3%83%BC%E3%83%88)を開始します。

1. サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png)**Workflows**をクリックし、**Delta Live Tables**タブをクリックします。**Pipelines**一覧が表示されます。
1. 以下のいずれかを行います。
    - パイプラインのアップデートをすぐにスタートするには、**Actions**カラムの![](https://docs.databricks.com/_images/right-arrow.png)をクリックします。システムがパイプラインを起動しているというメッセージを返します。
    - パイプラインをスタートする前にオプションを確認するには、パイプライン名をクリックします。**Pipeline details**ページが表示されます。

**Pipeline details**ページでは以下のオプションが提供されます。

- 開発モードとプロダクションモードを切り替えるために、![](https://docs.databricks.com/_images/dlt-env-toggle.png)ボタンを使用します。デフォルトではパイプラインは開発モードで実行されます。[開発、プロダクションモード](https://qiita.com/taka_yayoi/items/80ee29197d26e177bbeb#%E9%96%8B%E7%99%BA%E3%83%97%E3%83%AD%E3%83%80%E3%82%AF%E3%82%B7%E3%83%A7%E3%83%B3%E3%83%A2%E3%83%BC%E3%83%89)をご覧下さい。
- オプションとして、パイプラインのアクセス権を設定することができます。**Permissions**ボタンをクリックします。[Delta Live Tables access control](https://docs.databricks.com/security/access-control/dlt-acl.html)をご覧ください。
- パイプラインの設定を参照、編集するには**Settings**ボタンをクリックします。パイプライン設定の詳細については、[Delta Live Tablesの設定](https://qiita.com/taka_yayoi/items/834b56b30849d63b52d0)をご覧ください。

**Pipeline details**ページからパイプラインのアップデートをスタートするには、![](https://docs.databricks.com/_images/dlt-start-button.png)ボタンをクリックします。

例えば、新たな要件に基づいてクエリーを変更したり、新たなカラムを計算するバグを修正するなどのケースで、すでに取り込んだデータを再処理したい場合があるかもしれません。UIからフルリフレッシュの実行をDelta Live Tablesシステムに指示することで取り込み済みのデータを再処理することができます。フルリフレッシュを実行するには、**Start**ボタンの隣の![](https://docs.databricks.com/_images/down-caret-blue.png)をクリックして、**Full refresh all**をクリックします。

アップデートあるいはフルリフレッシュをスタートした後に、システムがパイプラインが起動中である旨のメッセージを返します。

アップデートの起動に成功すると、Delta Live Tablesは以下を実行します。

1. Delta Live Tablesシステムによって作成されたクラスター設定を用いてクラスターを起動します。また、カスタムの[クラスター設定](https://qiita.com/taka_yayoi/items/834b56b30849d63b52d0#%E3%82%AF%E3%83%A9%E3%82%B9%E3%82%BF%E3%83%BC%E8%A8%AD%E5%AE%9A)を指定することができます。
1. 存在しないテーブルを作成し、スキーマが既存テーブルに適合していることを確認します。
1. 最新のデータを用いてテーブルをアップデートします。
1. アップデートが完了したらクラスターをシャットダウンします。

**Pipeline details**ページの下部にあるイベントログを参照することでアップデートの進捗を追跡することができます。
![](https://docs.databricks.com/_images/dlt-sales-event-log.png)

ログエントリーの詳細を参照するにはエントリーをクリックします。**Pipeline event log details**ポップアップが表示されます。ログ詳細を含むJSONドキュメントを参照するには、**JSON**タブをクリックします。

例えば、パフォーマンス、データ品質メトリクスを解析するために、イベントログへのクエリーする方法は[Delta Live Tablesのイベントログ](https://qiita.com/taka_yayoi/items/5ae5824cfab30caf9795)をご覧ください。

パイプラインのアップデートが完了すると、選択したテーブルのみをリフレッシュするためのアップデートをスタートできるようになります。

# 選択したテーブルのパイプラインアップデートのスタート

パイプラインで選択したテーブルのみのデータを再処理したい場合があるかと思います。例えば、開発中に単一のテーブルのみを変更し、テストに要する時間を削減したい、あるいは、パイプラインのアップデートが失敗し、[失敗したテーブル](#失敗したテーブルのパイプラインのアップデートのスタート)のみをリフレッシュしたいと考えるかもしれません。

選択したテーブルのみをリフレッシュするアップデートをスタートするには、**Pipeline details**ページで以下のことを行います。

1. **Select tables for refresh**をクリックします。**Select tables for refresh**ダイアログが表示されます。
    **Select tables for refresh**ボタンが表示されない場合、**Pipeline details**ページには最新のアップデートが表示され、アップデートが完了していることを確認してください。最新のアップデートのDAGが表示されない場合、例えば、アップデートが失敗した場合には**Select tables for refresh**ボタンは表示されません。

1. リフレッシュするテーブルを選択するには、それぞれのテーブルをクリックします。選択したテーブルがハイライトされてラベルがつけられます。アップデート対象からテーブルを除外するには、再度テーブルをクリックします。
1. **Refresh selection**をクリックします。

    > **注意**
**Refresh selection**ボタンには、括弧内に選択したテーブルの数が表示されます。

選択されたテーブルにすでに取り込まれたデータを再処理するには、**Refresh selection**ボタンの隣の![](https://docs.databricks.com/_images/down-caret-blue.png)の**Full Refresh selection**ボタンをクリックします。

# 失敗したテーブルのパイプラインのアップデートのスタート

パイプライングラフの1つ以上のテーブルのエラーによってパイプラインのアップデートが失敗すると、失敗したテーブルと後段の依存関係のアップデートを起動することができます。

:::note info
**注意**
失敗したテーブルに依存していたとしても、除外されたテーブルはリフレッシュされません。
:::

失敗したテーブルをアップデートするには、**Pipeline details**ページで**Refresh failed tables**をクリックします。

選択した失敗テーブルのみをアップデートするには、

1. **Refresh failed tables**ボタンの隣の![](https://docs.databricks.com/_images/button-down.png)をクリックし、**Select tables for refresh**をクリックします。**Select tables for refresh**ダイアログが表示されます。
1. リフレッシュするテーブルを選択するには、それぞれのテーブルをクリックします。選択したテーブルはハイライトされ、ラベル付けされます。アップデートからテーブルを削除するには、再度テーブルをクリックします。
1. **Refresh selection**をクリックします。

    > **注意**
**Refresh selection**ボタンには、括弧内に選択したテーブルの数が表示されます。

選択したテーブルですでに取り込まれたデータを再処理するには、**Refresh selection**ボタンの隣の![](https://docs.databricks.com/_images/down-caret-blue.png)をクリックし、**Full Refresh selection**をクリックします。

# パイプライン詳細の参照

## パイプラインのグラフ

パイプラインの起動に成功すると、パイプラインのグラフが表示されます。マウスあるいは、グラフパネルの隅にある![](https://docs.databricks.com/_images/dlt-graph-buttons.png)を使用して表示を調整します。
![](https://docs.databricks.com/_images/dlt-sales-graph.png)

データ品質メトリクスのツールチップを表示するには、パイプライングラフのデータセットのデータ品質の値の上にマウスカーソルを移動します。

選択されたテーブルのみをリフレッシュするアップデートを実行する際、リフレッシュの対象ではないテーブルはパイプライングラフで**Excluded**トラベルがつきます。

## パイプラインの詳細

**Pipeline details**パネルには、パイプライン、アップデートのID、アップデートのステータス、アップデートのタイプ、アップデートのランタイムを含むパイプラインに関する情報と、パイプラインの最新のアップデートに関する情報が表示されます。

また、**Pipeline Details**パネルには、パイプラインに設定されている計算資源のコスト、製品エディション、Databricksランタイムバージョン、チャネルを含むパイプラインの計算クラスターに関する情報を表示します。クラスターのSpark UIが新規タブに表示されますので、**Spark UI**ボタンをクリックします。新規タブでクラスターログを開くには、**Logs**ボタンをクリックします。新規タブでクラスターのメトリクスを開くには、**Metrics**ボタンをクリックします。

**Run as**の値には、パイプラインのアップデートを実行するユーザーが表示されます。**Run as**ユーザーはパイプラインのオーナーであり、パイプラインのアップデートはこのユーザーのアクセス権でアップデートが実行されます。`run as`ユーザーを変更するには、**Permissions**をクリックし、パイプラインのオーナーを変更します。

## データセットの詳細

データセットのスキーマやデータ品質メトリクスを含むデータセットの詳細を表示するには、**Graph**ビューでデータセットをクリックします。データセット詳細が表示されます。
![](https://docs.databricks.com/_images/dlt-sales-dataset-details.png)

新規ウィンドウでパイプラインノートブックを開くには、**Path**の値をクリックします。

データセット詳細ビューを閉じて、**Pipeline details**に戻るには![](https://docs.databricks.com/_images/dlt-close-button.png)をクリックします。

# パイプラインアップデートの停止

パイプラインのアップデートを停止するには、![](https://docs.databricks.com/_images/dlt-stop-button.png)をクリックします。

# パイプラインのスケジュール

トリガーパイプラインを手動でスタート、あるいはDatabricks[ジョブ](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a)を用いてパイプラインをスケジュール実行することができます。Delta Live TablesのUIで直接単一のパイプラインタスクを用いたジョブを作成するか、ジョブのUIでマルチタスクワークフローにパイプラインタスクを追加することで、処理をスケジューリングすることができます。

単一タスクのジョブを作成し、Delta Live TablesのUIでジョブをスケジューリングするには以下を行います。

1. **Schedule > Add a schedule**をクリックします。パイプラインが1つ以上のスケジュールジョブに含まれている場合、**Schedule**ボタンには**Schedule (5)** というように既存のスケジュールの数が表示されます。
1. **Job name**フィールドにジョブの名前を入力します。
1. **Schedule**を**Scheduled**に設定します。
1. 期間、開始時刻、タイムゾーンを指定します。
1. パイプラインのスタート、成功、失敗時にアラートを受け取る1つ以上のメールアドレスを指定します。
1. **Create**をクリックします。

Databricksジョブでマルチタスクワークフローを作成し、パイプラインタスクを追加するには以下を行います。

1. ジョブのUIで[ジョブを作成](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**タスクを用いてジョブワークフローにあなたのパイプラインを追加します。
1. ジョブのUIでジョブの[スケジュールを作成](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E3%81%AE%E3%82%B9%E3%82%B1%E3%82%B8%E3%83%A5%E3%83%BC%E3%83%AB)します。

パイプラインのスケジュールを作成すると、以下のことが行えるようになります。

- Delta Live TablesのUIで、スケジュール名、一時停止されているかどうか、最後の実行時刻、最後の実行のステータスを含むスケジュールのサマリーの参照。
- [ジョブ](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E3%81%AE%E7%B7%A8%E9%9B%86)あるいは[パイプラインタスク](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%BF%E3%82%B9%E3%82%AF%E3%81%AE%E7%B7%A8%E9%9B%86)の編集。
- [スケジュール](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E3%81%AE%E3%82%B9%E3%82%B1%E3%82%B8%E3%83%A5%E3%83%BC%E3%83%AB)の編集、スケジュールの[停止及び再開](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E3%81%AE%E4%B8%80%E6%99%82%E5%81%9C%E6%AD%A2%E5%86%8D%E9%96%8B)。また、スケジュールを作成する際に**Manual**を選択した場合、スケジュールは停止状態となります。
- 手動によるジョブの[実行](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E3%81%AE%E5%AE%9F%E8%A1%8C)、ジョブの実行[詳細](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E5%AE%9F%E8%A1%8C%E3%81%AE%E5%8F%82%E7%85%A7)の参照。

# パイプラインの参照

サイドバーの![](https://docs.databricks.com/_images/jobs-icon.png)**Workflows**をクリックし、**Delta Live Tables**タブをクリックします。定義されたすべてのパイプラインの一覧、最近のパイプラインアップデートのステータス、パイプラインのID、パイプラインの作成者を表示する**Pipelines**ページが表示されます。

パイプライン一覧を以下の手段でフィルタリングすることができます。

- パイプライン名
- パイプライン名に部分一致するテキスト
- あなたが所有しているパイプラインのみ
- あなたがアクセスできるすべてのパイプライン

パイプラインの名前の昇順(A -> Z)、降順(Z -> A)でソートするには、**Name**カラムヘッダーをクリックします。

パイプライン一覧を参照する際には、パイプライン名はリンクになっているので、パイプライン詳細を新規タブ、新規ウィンドウで開くためには右クリックしてコンテキストメニューを使用することができます。

# 設定の編集

パイプラインの設定を参照、編集するには、**Pipeline details**ページで**Settings**ボタンをクリックします。設定を追加、変更、削除することができます。例えば、パイプラインを作成した後にパイプラインの出力をクエリーできるようにするには以下を行います。

1. **Settings**ボタンをクリックします。**Edit Pipeline Settings**ダイアログが表示されます。
1. **Target**フィールドにデータベース名を入力します。
1. **Save**をクリックします。

JSONの定義を参照、編集するには、**JSON**ボタンをクリックします。
![](https://docs.databricks.com/_images/dlt-settings-publish-sales.png)

設定の詳細については、[Delta Live Tablesの設定](https://qiita.com/taka_yayoi/items/834b56b30849d63b52d0)をご覧下さい。

# アップデート履歴の参照

[パイプラインのアップデート](https://qiita.com/taka_yayoi/items/80ee29197d26e177bbeb#%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%81%AE%E3%82%A2%E3%83%83%E3%83%97%E3%83%87%E3%83%BC%E3%83%88)の履歴とステータスを参照するには、**Update history**ドロップダウンをクリックします。
![](https://docs.databricks.com/_images/dlt-update-history.png)

アップデートのグラフ、詳細、イベントを参照するには、ドロップダウンのアップデートを選択します。最新のアップデートに戻るには、**Show the latest update**をクリックします。

# データセットの公開

パイプラインを作成、編集する際、Databricksの[メタストア](https://docs.databricks.com/data/metastores/index.html)にテーブル定義を公開し、レコードを[Delta](https://docs.databricks.com/delta/index.html)テーブルに永続化するために設定`target`を指定することができます。

アップデートが完了すると、[データベースとテーブルを参照](https://docs.databricks.com/data/data-tab.html#view-databases-and-tables)でき、データをクエリーし、後段のアプリケーションでデータを活用できるようになります。

詳細は[Delta Live Tables data publishing](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-publish.html)を参照ください。

# クラスターサイズの管理

パイプラインで使用するクラスターのリソースを管理することができます。デフォルトでは、Delta Live Tablesはパフォーマンスとコストを最適化するために、パイプラインのクラスターを自動でスケールします。クラスターのオートスケーリングを使用することをお勧めしますが、オプションとしてパイプラインを作成、編集する体に、オートスケーリングを無効化し、固定のワーカーノードの数を指定することができます。

- パイプラインを作成する際に**Enable autoscaling**チェックボックスを無効化し、**Workers**フィールドにノード数を指定します。
- 既存のパイプラインでオートスケーリングを無効化するために設定を変更します。以下のスニペットでは、クラスターのオートスケーリングが有効化される設定を示しています。

    ```json:JSON
    "clusters": [
       {
         "label": "default",
          "autoscale": {
            "min_workers": 1,
            "max_workers": 5
          }
       }
    ]
    ```

    以下のスニペットでは、オートスケーリングが無効化され、ワーカーノードの台数は5に固定されています。

    ```json:JSON
    "clusters": [
       {
         "label": "default",
         "num_workers": 5
       }
    ]
    ```

# パイプラインの削除

**Pipelines**一覧、あるいは**Pipeline details**ページでパイプラインを削除することができます。

- **Pipelines**一覧では**Actions**カラムの![](https://docs.databricks.com/_images/trash-icon-jobs.png)をクリックします。
- 対象のパイプラインの**Pipeline details**ページでは、**Delete**ボタンをクリックします。

パイプラインを削除することで、Delta Live Tablesのシステムからパイプライン定義が削除され、これを取り消すことはできません。

### Databricks 無料トライアル

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