Delta Live Tablesの設定

[Delta Live Tables settings \| Databricks on AWS](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-configuration.html) [2023/2/24時点]の翻訳です。

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

Delta Live Tablesの設定では、パイプラインを実装する一つ以上のノートブック、開発、ステージング、プロダクション環境でパイプラインをどのように実行するのかを設定するパラメーターを指定します。Delta Live Tablesの設定はJSONで表現され、[Delta Live TablesのUI](https://qiita.com/taka_yayoi/items/7fe8ed2c2f95fd53cc3d#%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E8%A8%AD%E5%AE%9A%E3%81%AE%E5%8F%82%E7%85%A7)で編集することができます。


# 設定

| フィールド |
|:--|
|**id**<br>Type:`string`<br>グローバルでユニークなパイプラインのID。IDはシステムによって割り当てられ変更することはできません。   |
|**name**<br>Type:`string`<br>わかりやすいパイプラインの名称。UIでパイプラインジョブを識別するために使用されます。   |
|**storage**<br>Type:`string`<br>パイプラインの実行に必要な出力データ、メタデータを格納するDBFSあるいはクラウドストレージ。<br>`storage`が設定されない場合、システムがデフォルトの`dbfs:/pipelines/`を使用します。<br>パイプラインの作成後に`storage`を変更することはできません。  |
|**configuration**<br>Type:`object `<br>パイプラインを実行するクラスターのSpark設定を追加するためのオプションの設定リストです。これらの設定はDelta Live Tablesランタイムに読み込まれ、Spark設定を通じてパイプラインクエリーで使用されます。<br>各々の要素は`key:value`の形式である必要があります。<br>`configuration`の使用例については、[パイプラインのパラメータ化](#パイプラインのパラメータ化)を参照ください。   |
|**libraries**<br>Type:`array of objects`<br>パイプラインのコードと必要とされるアーティファクトを含むノートブックの配列を指定します。サンプルは[パイプラインの複数のノートブックの設定](#パイプラインの複数のノートブックの設定)を参照ください。   |
|**clusters**<br>Type:`array of objects`<br>パイプラインを実行するクラスターの設定の配列です。詳細は[クラスター設定](#クラスター設定)を参照ください。   |
|**development**<br>Type:`boolean`<br>パイプラインを`development`あるいは`production`モードで実行するかを示すフラグです。[開発、プロダクションモード](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)をご覧ください。   |
|**continuous**<br>Type:`boolean`<br>パイプラインを連続で稼働させるかどうかを示すフラグです。<br>デフォルトは`false`です。   |
|**target**<br>Type:`string`<br>パイプラインの出力データを永続化するデータベース名です。`target`を設定する事で、DatabricksのUIからパイプラインの出力データを参照、検索することが可能となります。   |
|**channel**<br>Type:`string`<br>使用するDelta Live Tablesランタイムのバージョンです。サポートされる値は:<ul><li>ランタイムバージョンに間も無く加えられる変更をパイプラインでテストするには`preview`を指定します。<li>現在のランタイムバージョンを指定するには`current`を指定します。</ul>`channel`フィールドはオプションです。デフォルトは`current`です。プロダクションのワークロードでは`current`バージョンのランタイムを使用することをお勧めします。   |
|**edition**<br>Type:`string`<br>パイプラインを実行する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)です。この設定によって、パイプラインの要件に応じたベストな製品エディションを選択することができます。<ul><li>ストリーミング取り込みワークロードを実行するには`CORE`を使用します。<li>ストリーミング取り込みとチェンジデータキャプチャ(CDC)ワークロードを実行するには`PRO`を使用します。<li>ストリーミング取り込みとチェンジデータキャプチャ(CDC)ワークロード、データ品質制約を強制するためのDelta Live Tablesエクスペクテーションを使用するには`ADVANCED`を使用します。</ul>`edition`フィールドはオプションです。デフォルトは`ADVANCED`です。   |
|**photon**<br>Type:`boolean`<br>パイプラインを実行するために[Photonランタイム](https://docs.databricks.com/runtime/photon.html)を使用するかどうかを指定するフラグです。PhotonはDatabricksの高性能Sparkエンジンです。Photonが有効化されたパイプラインの課金と、有効化されていないパイプラインの課金は異なります。<br>`photon`フィールドはオプションです。デフォルトは`false`です。   |
|**pipelines.maxFlowRetryAttempts**<br>Type:`int`<br>リトライ可能な障害が発生した際、パイプラインのアップデートが失敗する前にフローのリトライを試みる最大数です。<br>デフォルト値は2です。デフォルトでは、リトライ可能な障害が発生した場合、Delta Live Tablesランタイムは、初回の試行を含み3回フローの実行を試みます。   |
|**pipelines.numUpdateRetryAttempts**<br>Type:`int`<br>リトライ可能な障害が発生した際、パイプラインのアップデートが失敗する前にアップデートのリトライを試みる最大数です。リトライはフルアップデートとして実行されます。<br>デフォルト値は5です。このパラメーターはproductionモードのアップデート実行の際にのみ適用されます。パイプライン実行がdevelopmentモードの場合にはリトライは行われません。   |

# パイプラインのトリガー間隔

テーブルをアップデートするフロー、あるいはパイプライン全体のトリガー間隔をコントロールするために、`pipelines.trigger.interval`を活用することができます。起動されるパイプラインはそれぞれのテーブルを一度のみ処理するので、`pipelines.trigger.interval`は[連続パイプライン](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)の場合にのみ使用されます。

ストリーミングクエリーとバッチクエリーとでは異なるデフォルト値があるため、個々のテーブルで`pipelines.trigger.interval`を設定することをお勧めします。処理においてパイプラインのグラフ全体のアップデートをコントロールする必要がある場合にのみ、パイプラインにこの値を設定してください。

Pythonの`spark_conf`、SQLの`SET`を用いてテーブルに`pipelines.trigger.interval`を設定します。

```py:Python
@dlt.table(
  spark_conf={"pipelines.trigger.interval" : "10 seconds"}
)
def <function-name>():
    return (<query>)
```

```sql:SQL
SET pipelines.trigger.interval='10 seconds';

CREATE OR REFRESH LIVE TABLE TABLE_NAME
AS SELECT ...
```

パイプラインに`pipelines.trigger.interval`を設定するには、パイプライン設定の`configuration`オブジェクトに追加します。

```json:JSON
{
  "configuration": {
    "pipelines.trigger.interval": "10 seconds"
  }
}
```

| pipelines.trigger.interval|
|:--|
| デフォルト値はフロータイプによって変わります:<ul><li>ストリーミングクエリーは5秒<li>すべての入力データがDeltaソースのコンプリートクエリーは1分<li>いくつかのデータソースが非Deltaの古プリートクエリーは10分。[連続パイプラインにおけるテーブルとビュー](https://qiita.com/taka_yayoi/items/80ee29197d26e177bbeb#%E9%80%A3%E7%B6%9A%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%81%AB%E3%81%8A%E3%81%91%E3%82%8B%E3%83%86%E3%83%BC%E3%83%96%E3%83%AB%E3%81%A8%E3%83%93%E3%83%A5%E3%83%BC)をご覧ください。</ul>値は数値と時間単位となります。以下の例は適切な時間単位です:<ul><li>`second`, `seconds`<li>`minute`, `minutes`<li>`hour`, `hours`<li>`day`, `days`</ul>以下のように値を定義する際に単数形、複数形を使用することができます。<ul><li>`{"pipelines.trigger.interval" : "1 hour"}`<li>`{"pipelines.trigger.interval" : "10 seconds"}`<li>`{"pipelines.trigger.interval" : "30 second"}`<li>`"pipelines.trigger.interval" : "1 minute"}`<li>`{"pipelines.trigger.interval" : "10 minutes"}`<li>`{"pipelines.trigger.interval" : "10 minute"}`</ul>  |


# クラスター設定

[Create cluster](https://docs.databricks.com/dev-tools/api/latest/clusters.html#create) APIと同じJSONフォーマットを用いて、パイプラインで使用されるクラスターを設定することができます。二種類のクラスタータイプに対する設定を行うことができます。全ての処理が行われる*default*クラスターと、日次のメンテナンスタスクを実行する*maintenance*クラスターです。それぞれのクラスターは`label`フィールドで識別することができます。

クラスターのプロパティの指定はオプションであり、指定されない値に関してはシステムはデフォルト値を使用します。

:::note
**注意**
- クラスター設定でSparkのバージョンを指定することはできません。Delta Live Tablesのクラスターは、最新機能を取り込むように定期的にアップデートされるDatabricksランタイムのカスタムバージョンで動作します。マニュアルでバージョンを設定するとパイプラインが失敗する場合があります。
- Delta Live Tablesのクラスターは使用されていない時に自動でシャットダウンされるため、クラスター設定の`autotermination_minutes`を設定するクラスターポリシーの参照はエラーとなります。クラスターのシャットダウンの挙動を制御するために、[developmentモードかproductionモード](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)を使用するか、Spark設定`pipelines.clusterShutdown.delay`を使用することができます。以下の例では、`pipelines.clusterShutdown.delay`を5秒に設定しています。

    ```json:JSON
    {
      "configuration": {
      "pipelines.clusterShutdown.delay": "60s"
    }
    ```
    `production`モードが有効化されている場合、`pipelines.clusterShutdown.delay`のデフォルト値は`0 seconds`です。`development`モードが有効化されている場合、デフォルト値は`2 hours`になります。
- クラスター設定で`num_workers`を0に設定すると、クラスターは[シングルノードクラスター](https://qiita.com/taka_yayoi/items/6e7c6cc14a5017ef6d15)として作成されます。オートスケーリングクラスターを設定し、`min_workers`を0、`max_workers`を0に設定してもシングルノードクラスターが作成されます。

    オートスケーリングクラスターを設定し、`min_workers`のみを0に設定すると、クラスターはシングルノードクラスターとしては作成されません。このクラスターは停止されるまで常に1台のワーカーが存在することになります。

    Delta Live Tablesでシングルノードクラスターを作成するクラスター設定のサンプルです。

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

:::

:::note
**注意**
お使いのストレージにアクセスするためにインスタンスプロファイルや他の設定が必要な場合には、defaultクラスター、maintenanceクラスターの両方に設定してください。
:::

defaultクラスター、maintenanceクラスターの設定例を以下に示します。

```json:JSON
{
  "clusters": [
    {
      "label": "default",
      "node_type_id": "c5.4xlarge",
      "driver_node_type_id": "c5.4xlarge",
      "num_workers": 20,
      "spark_conf": {
        "spark.databricks.io.parquet.nativeReader.enabled": "false"
      },
      "aws_attributes": {
        "instance_profile_arn": "arn:aws:..."
      }
    },
    {
      "label": "maintenance",
      "aws_attributes": {
        "instance_profile_arn": "arn:aws:..."
      }
    }
  ]
}
```

## クラウドストレージの設定

[AWSのS3ストレージ](https://qiita.com/taka_yayoi/items/446c7971be354f88c679)へのアクセスを設定するには、AWSインスタンスプロファイルを使用します。Delta Live Tables APIあるいはDelta Live Tables UIでパイプラインを[作成](https://docs.databricks.com/workflows/delta-live-tables/delta-live-tables-api-guide.html#dlt-create-pipeline)、[編集](https://docs.databricks.com/workflows/delta-live-tables/delta-live-tables-api-guide.html#dlt-edit-pipeline)する際にインスタンスプロファイルの設定を追加することができます。また、パイプラインクラスターのJSON設定を直接編集することでAWSインスタンスプロファイルを設定することができます。

1. パイプラインの**Pipeline details**ページで、**Settings**ボタンをクリックします。**Pipeline settings**ページが表示されます。
1. **JSON**ボタンをクリックします。
1. クラスター設定の`aws_attributes.instance_profile_arn`フィールドに、インスタンスプロファイル設定を入力します。

```json:JSON
{
  "clusters": [
    {
      "label": "default",
      "aws_attributes": {
        "instance_profile_arn": "arn:aws:..."
      }
    },
    {
      "label": "maintenance",
      "aws_attributes": {
        "instance_profile_arn": "arn:aws:..."
      }
    }
  ]
}
```

defaultとmaintenanceクラスターに対してインスタンスプロファイルを指定する必要があります。

Delta Live Tablesパイプラインの[クラスターポリシー](#クラスターポリシー)を作成する際に、インスタンスプロファイルを設定することもできます。サンプルに関しては、[knowledge base](https://kb.databricks.com/clusters/set-instance-profile-arn-optional-policy)をご覧ください。

## クラスターポリシー

:::note
**注意**
Delta Live Tablesクラスターを設定するためにクラスターポリシーを使う際には、`default`と`maintenance`クラスターの両方で単一のポリシーを適用することをお勧めします。
:::

パイプラインクラスターで[クラスターポリシー](https://qiita.com/taka_yayoi/items/93d7597d7fc739b48bb7)を設定するには、`cluster_type`フィールドを`dlt`に設定したポリシーを作成します。

```json:JSON
{
  "cluster_type": {
    "type": "fixed",
    "value": "dlt"
  }
}
```

パイプライン設定で、clusterの`policy_id`フィールドをポリシーIDの値に設定します。以下の例では、ID`C65B864F02000008`のクラスターポリシーを用いて`default`と`maintenance`クラスターを設定します。

```json:JSON
{
  "clusters": [
    {
      "label": "default",
      "policy_id": "C65B864F02000008",
      "autoscale": {
        "min_workers": 1,
        "max_workers": 5,
        "mode": "ENHANCED"
      }
    },
    {
      "label": "maintenance",
      "policy_id": "C65B864F02000008"
    }
  ]
}
```

クラスターポリシーの作成、使用のサンプルについては、[Define limits on pipeline clusters](https://docs.databricks.com/workflows/delta-live-tables/delta-live-tables-cookbook.html#use-cluster-policies)をご覧ください。

# サンプル

## パイプラインとクラスターの設定

以下の例では、ストレージとしてのDBFS、小規模の1ノードのクラスターを使用して、**example-notebook_1**で実装されたパイプラインを実行するトリガーモードのパイプラインを設定しています。

```json:JSON
{
  "name": "Example pipeline 1",
  "storage": "dbfs:/pipeline-examples/storage-location/example1",
  "clusters": [
    {
      "num_workers": 1,
      "spark_conf": {}
    }
  ],
  "libraries": [
    {
      "notebook": {
         "path": "/Users/user@databricks.com/example_notebook_1"
      }
    }
  ],
  "continuous": false
}
```

## パイプラインの複数のノートブックの設定

複数のノートブックを用いたパイプラインを設定するには`libraries`フィールドを使用します。Delta Live Tablesは、パイプラインに対する処理グラフを構築するためにデータセットの依存関係を自動で解析するので、ノートブックの順序は問いません。以下の例では、**example-notebook_1**と**example-notebook_2**で定義されるデータセットを含むパイプラインを作成します。

```json:JSON
{
  "name": "Example pipeline 3",
  "storage": "dbfs:/pipeline-examples/storage-location/example3",
  "libraries": [
    { "notebook": { "path": "/example-notebook_1" } },
    { "notebook": { "path": "/example-notebook_2" } }
  ]
}
```

## ワークスペースファイルからコードをインポート

パイプラインコードを含むノートブックを設定するために`libraries`フィールドを用いることに加え、Databricks repoに格納されているソースコードに含まれるパイプラインコードを設定することができます。パイプラインコードを含むソースファイルへのパスを指定するには、`file`フィールドを使用します。

```json:JSON
{
   "libraries": [
     { "file": { "path": "/Repos/<user_name>@databricks.com/Apply_Changes_Into/apply_changes_into.sql" } },
     { "file": { "path": "/Repos/<user_name>@databricks.com/Apply_Changes_Into/apply_changes_into.py" } }
   ]
}
```

## Delta Live Tablesによる開発ワークフローの作成

開発、ステージング、プロダクションに対応する別々のDelta Live Tablesパイプラインを作成することができるので、作成するデータのコンシューマーに影響を与えることなしに、変換ロジックをテスト、デバッグすることができます。異なるデータベースをターゲットとしつつも同じコードを使用する別々のパイプラインを作成するだけです。

完全に分離された開発環境と、開発環境からプロダクション環境にプッシュするシンプルなワークフローを構築するために[DatabricksのRepos](https://qiita.com/taka_yayoi/items/b89f199ff0d3a4c16140)の機能と組み合わせることができます。

```json:JSON
{
  "name": "Data Ingest - DEV user@databricks",
  "target": "customers_dev_user",
  "libraries": ["/Repos/user@databricks.com/ingestion/etl.py"],
}
```

```json:JSON
{
  "name": "Data Ingest - PROD",
  "target": "customers",
  "libraries": ["/Repos/production/ingestion/etl.py"],
}
```

## パイプラインのパラメータ化

データセットを定義するPython、SQLコードはパイプラインの設定でパラメータ化することができます。パラメータ化を用いることで以下のユースケースを実現することができます。

- コードと長いパスや変数の分離
- 開発、ステージング環境で処理するデータ量を削減し、テストを高速化
- 複数のデータソースに対する処理で同じ変換ロジックを再利用

以下の例では、開発用パイプラインで使用するデータを入力データのサブセットに限定するために、設定値`startDate`を使用しています。

```sql:SQL
CREATE LIVE TABLE customer_events
AS SELECT * FROM sourceTable WHERE date > ${mypipeline.startDate};
```

```py:Python
@dlt.table
def customer_events():
  start_date = spark.conf.get("mypipeline.startDate")
  return read("sourceTable").where(col("date") > start_date)
```

```json:JSON
{
  "name": "Data Ingest - DEV",
  "configuration": {
    "mypipeline.startDate": "2021-01-02"
  }
}
```

```json:JSON
{
  "name": "Data Ingest - PROD",
  "configuration": {
    "mypipeline.startDate": "2010-01-02"
  }
}
```

### Databricks 無料トライアル

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