Delta Live Tablesによるチェンジデータキャプチャ(CDC)

[Change data capture with Delta Live Tables \| Databricks on AWS](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-cdc.html) [2022/8/11時点]の翻訳です。

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

:::note
**注意**
本書では、ソースデータの変更に基づいて、どのようにDelta Live Tablesのパイプラインをアップデートするのかを説明します。Deltaテーブルにおける行レベルの変更を記録しクエリーする方法については、[チェンジデータフィード](https://qiita.com/taka_yayoi/items/e3c40fce43561f73a50c)をご覧ください。
:::

:::note
**プレビュー**
Delta Live TablesにおけるSCD type 2サポートは[Public Preview](https://docs.databricks.com/release-notes/release-types.html)です。
:::

ソースデータの変更に基づいてテーブルをアップデートするために、Delta Live Tablesでチェンジデータキャプチャ(CDC)を利用することができます。CDCはDelta Live TablesのSQLおよびPythonインタフェースでサポートされています。また、Delta Live Tablesでは、slowly changing dimensions (SCD)のタイプ1とタイプ2をサポートしています。

- レコードを直接更新するにはSCDタイプ1を使用します。更新されたレコードに対する履歴は保持されません。
- レコードのすべての更新履歴をを保持するにはSCDタイプ2を使用します。

変更の有効期間を表現するために、SCDタイプ2では、すべての変更を生成された`__START_AT`と`__END_AT`カラムと共に格納します。`__START_AT`と`__END_AT`カラムを生成するために、Delta Live TablesはSQLでは`SEQUENCE BY`、Pythonでは`sequence_by`で指定されるカラムを使用します。

:::note
**注意**
`__START_AT`と`__END_AT`カラムのデータ型は`SEQUENCE BY`フィールドで指定されたデータ型と同じものになります。
:::

# SQL

Delta Live Tables CDCの機能を使用するためには、`APPLY CHANGES INTO`文を使用します。

```sql:SQL
APPLY CHANGES INTO LIVE.table_name
FROM source
KEYS (keys)
[WHERE condition]
[IGNORE NULL UPDATES]
[APPLY AS DELETE WHEN condition]
[APPLY AS TRUNCATE WHEN condition]
SEQUENCE BY orderByColumn
[COLUMNS {columnList | * EXCEPT (exceptColumnList)}]
[STORED AS {SCD TYPE 1 | SCD TYPE 2}]
```

| 句 |
|:--|
|**KEYS**<br><br>ソースデータで行を一意に特定するカラム、あるいはカラムの組み合わせです。これは、ターゲットテーブルの特定のレコードに適用されるCDCイベントの特定に使用されます。<br><br>この句は必須です。   |
|**WHERE**<br><br>パーティションプルーニングのような最適化処理を起動するために、ソース、ターゲットテーブルの両方に適用される条件です。この条件はソースのレコードを削除することに使用することはできません。ソーステーブルのすべてのCDCレコードはこの条件を満足する必要があり、そうでない場合にはエラーが発生します。WHERE句の使用は任意であり、お使いの処理で特定の最適化が必要な場合に使用されるべきです。<br><br>この句は任意です。   |
|**IGNORE NULL UPDATES**<br><br>ターゲットカラムのサブセットに対する取り込み更新を許可します。CDCイベントが既存のレコードと合致し、IGNORE NULL UPDATESが指定されている場合、`null`を含むカラムは、ターゲットにおける既存の値を維持します。これは`null`値を持つネストされたカラムにも適用されます。<br><br>この句は任意です。<br><br>デフォルトでは、`null`値を持つ既存カラムを上書きします。   |
|**APPLY AS DELETE WHEN**<br><br>CDCイベントがupsertではなく`DELETE`として取り扱われるべきタイミングを指定します。out-of-orderのデータを取り扱うために、削除されたレコードは背後のDeltaテーブルで一時的にtombstoneとして保持され、これらのtombstoneを表示するビューがメタストアに作成されます。保持期間は[テーブルプロパティ](#テーブルプロパティ)の`pipelines.cdc.tombstoneGCThresholdInSeconds`で設定することができます。<br><br>この句は任意です。  |
|**APPLY AS TRUNCATE WHEN**<br><br>CDCイベントがテーブルの完全な`TRUNCATE`として取り扱われるべきタイミングを指定します。この句はターゲットテーブルを完全にtruncateするので、この機能を必要とする特定のユースケースでのみ使用すべきです。<br><br>`APPLY AS TRUNCATE WHEN`句はSCDタイプ1でのみサポートされます。SCDタイプ2ではtruncateをサポートしていません。<br><br>この句は任意です。  |
|**SEQUENCE BY**<br><br>ソースデータにおけるCDCイベントの論理的順序を指定するカラム名です。Delta Live Tablesは、out-of-orderな変更イベントを取り扱うために、この順序を使用します。<br><br>この句は必須です。   |
|**COLUMNS**<br><br>ターゲットテーブルに含めるカラムのサブセットを指定します。以下のいずれかが可能です。<ul><li>含めるカラムの完全なリストを指定：`COLUMNS (userId, name, city)`<li>除外するカラムのリストを指定：`COLUMNS * EXCEPT (operation, sequenceNum)`</ul>この句は任意です。<br><br>デフォルトでは、`COLUMNS`句が指定されない場合、ターゲットテーブルには全てのカラムが含まれます。   |
|**STORED AS**<br><br>レコードをSCDタイプ1あるいはSCDタイプ2で格納するのかを指定します。<br><br>この句は任意です。<br><br>デフォルトはSCDタイプ1です。   |

`INSERT`、`UPDATE`イベントのデフォルトの挙動は、ソースからのCDCイベントを*upsert*するというものです：指定されたキーにマッチするターゲットテーブルの行を更新する、あるいはターゲットテーブルにマッチする行がない場合には行を挿入します。`DELETE`イベントの挙動を`APPLY AS DELETE WHEN`条件で指定することも可能です。

# Python

Delta Live Tables CDC機能を使うためには、Python APIの`apply_changes()`関数を使用します。また、Delta Live TablesのPython CDCインタフェースは

## Apply changes関数

```py:Python
apply_changes(
  target = "<target-table>",
  source = "<data-source>",
  keys = ["key1", "key2", "keyN"],
  sequence_by = "<sequence-column>",
  ignore_null_updates = False,
  apply_as_deletes = None,
  apply_as_truncates = None,
  column_list = None,
  except_column_list = None,
  stored_as_scd_type = <type>
)
```

| 引数 |
|:--|
|**target**<br>Type:`str`<br>更新されるテーブル名です。<br>このパラメーターは必須です。   |
|**source**<br>Type:`str`<br>CDCレコードを含むデータソースです。<br>このパラメーターは必須です。   |
|**keys**<br>Type:`list`<br>ソースデータで行を一意に特定するカラム、あるいはカラムの組み合わせです。これは、ターゲットテーブルの特定のレコードに適用されるCDCイベントの特定に使用されます。<br><br>以下のいずれかの方法で指定できます：<ul><li>文字列のリスト：`["userId", "orderId"]`<li>Spark SQLの`col`関数のリスト：`[col("userId"), col("orderId"]`</ul>`col()`関数の引数にはクオリファイアを含めることはできません。例えば、`col(userId)`は使えますが、`col(source.userId)`は使えません。<br><br>このパラメーターは必須です。   |
|**sequence_by**<br>Type:`str`あるいは`col()`<br>ソースデータにおけるCDCイベントの論理的順序を指定するカラム名です。Delta Live Tablesは、out-of-orderな変更イベントを取り扱うために、この順序を使用します。<br><br>以下のいずれかの方法で指定できます：<ul><li>文字列：`"sequenceNum"`<li>Spark SQLの`col`関数：`col("sequenceNum")`</ul>`col()`関数の引数にはクオリファイアを含めることはできません。例えば、`col(userId)`は使えますが、`col(source.userId)`は使えません。<br><br>このパラメーターは必須です。   |
|**ignore_null_updates**<br>Type:`bool`<br>ターゲットカラムのサブセットに対する取り込み更新を許可します。CDCイベントが既存のレコードと合致し、`ignore_null_updates`が`True`の場合、`null`を含むカラムは、ターゲットにおける既存の値を維持します。これは`null`値を持つネストされたカラムにも適用されます。`ignore_null_updates`が`False`の場合、既存の値は`null`で上書きされます。<br><br>このパラメーターは任意です。 <br><br>デフォルトは`False`です。   |
|**apply_as_delete**<br>Type:`str`あるいは`expr()`<br>CDCイベントがupsertではなく`DELETE`として取り扱われるべきタイミングを指定します。out-of-orderのデータを取り扱うために、削除されたレコードは背後のDeltaテーブルで一時的にtombstoneとして保持され、これらのtombstoneを表示するビューがメタストアに作成されます。保持期間は[テーブルプロパティ](#テーブルプロパティ)の`pipelines.cdc.tombstoneGCThresholdInSeconds`で設定することができます。<br><br>以下のいずれかの方法で指定できます。<ul><li>文字列：`"Operation = 'DELETE'"`<li>Spark SQLの`expr()`関数：`expr("Operation = 'DELETE'")`</ul>このパラメーターは任意です。  |
|**apply_as_truncates**<br>Type:`str`あるいは`expr()`<br>CDCイベントがテーブルの完全な`TRUNCATE`として取り扱われるべきタイミングを指定します。この句はターゲットテーブルを完全にtruncateするので、この機能を必要とする特定のユースケースでのみ使用すべきです。<br><br>`apply_as_truncates`パラメータはSCDタイプ1でのみサポートされます。SCDタイプ2ではtruncateをサポートしていません。<br><br>以下のいずれかの方法で指定できます。<ul><li>文字列：`"Operation = 'TRUNCATE'"`<li>Spark SQLの`expr()`関数：`expr("Operation = 'TRUNCATE'")`</ul>このパラメーターは任意です。  |
|**column_list except_column_list**<br>Type:`list`<br>ターゲットテーブルに含めるカラムのサブセットを指定します。含めるカラムの完全なリストを指定するには`column_list`を使用します。除外するカラムを指定するには`except_column_list`を使用します。文字列のリストあるいはSpark SQLの`col()`関数で値を宣言することができます<ul><li>`column_list = ["userId", "name", "city"]`<li>`column_list = [col("userId"), col("name"), col("city")]`<li>`except_column_list = ["operation", "sequenceNum"]`<li>`except_column_list = [col("operation"), col("sequenceNum")`</ul>`col("sequenceNum")`</ul>`col()`関数の引数にはクオリファイアを含めることはできません。例えば、`col(userId)`は使えますが、`col(source.userId)`は使えません。<br><br>このパラメーターは任意です。<br><br>`column_list`や`except_column_list`引数が関数に渡されていない場合、デフォルトではターゲットテーブルの全てのカラムが含まれます。   |
|**stored_as_scd_type**<br>Type:`str`か`int`<br>レコードをSCDタイプ1あるいはSCDタイプ2で格納するのかを指定します。<br><br>SCDタイプ1では`1`、SCDタイプ2では`2`を指定します。<br><br>この句は任意です。<br><br>デフォルトはSCDタイプ1です。   |

`INSERT`、`UPDATE`イベントのデフォルトの挙動は、ソースからのCDCイベントを*upsert*するというものです：指定されたキーにマッチするターゲットテーブルの行を更新する、あるいはターゲットテーブルにマッチする行がない場合には行を挿入します。`DELETE`イベントの挙動を引数`apply_as_delete`で指定することも可能です。

## 出力レコードのためのターゲットテーブルの作成

`apply_changes()`の出力レコードのためのターゲットテーブルを作成するために`create_streaming_live_table()`関数を使用します。

:::note info
**注意**
`create_target_table()`関数は非推奨となります。既存のコードでは`create_streaming_live_table()`を使用するように更新することをお勧めします。
:::

```py:Python
create_streaming_live_table(
  name = "<table-name>",
  comment = "<comment>"
  spark_conf={"<key>" : "<value", "<key" : "<value>"},
  table_properties={"<key>" : "<value>", "<key>" : "<value>"},
  partition_cols=["<partition-column>", "<partition-column>"],
  path="<storage-location-path>",
  schema="schema-definition"
)
```

| 引数 |
|:--|
|**name**<br>Type:`str`<br><br>テーブル名<br><br>このパラメータは必須です。|
|**comment**<br>Type:`str`<br><br>テーブルに対するオプションのコメント。|
|**spark_conf**<br>Type:`dict`<br><br>このクエリーの実行におけるSpark設定のリスト。オプション。|
|**table_properties**<br>Type:`dict`<br><br>[テーブルのプロパティ](https://qiita.com/taka_yayoi/items/415687fa84e59de7a370#%E3%83%86%E3%83%BC%E3%83%96%E3%83%AB%E3%83%97%E3%83%AD%E3%83%91%E3%83%86%E3%82%A3)のリスト。オプション。|
|**partition_cols**<br>Type:`array`<br><br>テーブルをパーティショニングするための1つ以上のカラムのリスト。オプション。|
|**path**<br>Type:`str`<br><br>テーブルデータの格納場所。オプション。指定されない場合、パイプラインのストレージ場所をデフォルトとします。|
|**schema**<br>Type:`str`<br><br>テーブルのスキーマ定義。オプション。スキーマはSQL DDL文字列、Pythonの`StructType`で定義することができます。|

`apply_changes`のターゲットテーブルのスキーマを指定する際、`sequence_by`フィールドと同じデータ型を`__START_AT`と`__END_AT`カラムに指定して含める必要があります。例えば、ターゲットテーブルに`key, STRING, value, STRING sequencing, LONG`のカラムがある場合、以下のようになります。

```py:Python
create_streaming_live_table(
  name = "target",
  comment = "Target for CDC ingestion.",
  partition_cols=["value"],
  path="$tablePath",
  schema=
    StructType(
      [
        StructField('key', StringType()),
        StructField('value', StringType()),
        StructField('sequencing', LongType()),
        StructField('__START_AT', LongType()),
        StructField('__END_AT', LongType())
      ]
    )
)
```

:::note
**注意**
- `APPLY CHANGES INTO`クエリーや`apply_changes`関数を実行する前に、ターゲットテーブルが作成されていることを確認してください。[サンプルクエリー](#クエリー)を参照ください。
- 出力行数などターゲットテーブルのメトリクスは利用できません。
- SCDタイプ2は、カラムに変更がなくても入力行ごとに履歴を追加します。
- `APPLY CHANGES INTO`のクエリーや`apply_changes`関数のターゲットは、ストリーミングライブテーブルのソースとして使用することはできません。`APPLY CHANGES INTO`のクエリーや`apply_changes`関数のターゲットから読み込むテーブルはライブテーブルでなくてはいけません。
- `APPLY CHANGES INTO`クエリーや`apply_changes`関数ではエクスペクテーションはサポートされていません。ソース、ターゲットデータセットにエクスペクテーションを適用するには、以下の手順を踏んでください。
    - 必要なエクスペクテーションを持つ中間テーブルを定義することでソースデータにエクスペクテーションを追加し、このデータセットをターゲットテーブルに対するソースとします。
    - ターゲットテーブルから入力データを読み込む下流のテーブルを用いてターゲットデータに対するエクスペクテーションを追加します。
:::

# テーブルプロパティ

以下のテーブルプロパティは、`DELETE`イベントにおけるtombstone管理を制御するために追加されています。

| テーブルプロパティ |
|:--|
|**pipelines.cdc.tombstoneGCThresholdInSeconds**<br>out-of-orderのデータに期待する最大の間隔にマッチする値を設定します。   |
|**pipelines.cdc.tombstoneGCFrequencyInSeconds**<br>tombstoneとチェックされたデータのクリーンアップの頻度を指定します。<br>デフォルト：5分   |

# サンプル

このサンプルでは、以下のソースイベントに基づいてターゲットテーブルをアップデートするDelta Live TablesのSCDタイプ1とタイプ2のクエリーをデモします。

1. 新規ユーザーのレコードを作成
1. ユーザーレコードを削除
1. ユーザーレコードを更新。SCDタイプ1のサンプルでは、out-of-orderのイベントの取り扱いをデモするために、最後の`UPDATE`オペレーションは遅れて到着したので、ターゲットテーブルからは除外されます。

| userId | name | city | operation | sequenceNum |
|:--|:--|:--|:--|:--|
| 124  |  Raul |  Oaxaca | INSERT  | 1  |
| 123  |  Isabel | Monterrey  | INSERT  | 1  |
| 125  |  Mercedes | Tijuana  | INSERT   | 2  |
| 126  |  Lily | Cancun  | INSERT  | 2  |
| 123  |  null | null  | DELETE  | 6  |
| 125   | Mercedes  | Guadalajara  | UPDATE  |  	6 |
| 125  | Mercedes  |  Mexicali | UPDATE  |  5 |
| 123  |  Isabel |  Chihuahua |  UPDATE |  5 |

このSCDタイプ1のサンプルのクエリーを実行した後では、ターゲットテーブルには以下のレコードが含まれることになります。

| userId | name | city |
|:--|:--|:--|
| 124  | Raul  | Oaxaca  |
|  125 | Mercedes  | Guadalajara  |
|  126 | Lily  | Cancun  |

以下の入力レコードには、`TRUNCATE`オペレーションを伴う追加レコードが含まれており、SCDタイプ1のサンプルコードで使用することができます。

| userId | name | city | operation | sequenceNum |
|:--|:--|:--|:--|:--|
| 124  |  Raul |  Oaxaca | INSERT  | 1  |
| 123  |  Isabel | Monterrey  | INSERT  | 1  |
| 125  |  Mercedes | Tijuana  | INSERT   | 2  |
| 126  |  Lily | Cancun  | INSERT  | 2  |
| 123  |  null | null  | DELETE  | 6  |
| 125   | Mercedes  | Guadalajara  | UPDATE  |  	6 |
| 125  | Mercedes  |  Mexicali | UPDATE  |  5 |
| 123  |  Isabel |  Chihuahua |  UPDATE |  5 |
| null  |  null |  null |  TRUNCATE |  3 |

追加の`TRUNCATE`レコードを伴うSCDタイプ1のサンプルの実行後は、`TRUNCATE`オペレーションが`sequenceNum=3`であるため、レコード`124`と`125`は切り取られ、ターゲットテーブルには以下のレコードが含まれることになります。

| userId | name | city |
|:--|:--|:--|
| 125  | Mercedes  | Guadalajara  |

SCDタイプ2のサンプルの実行後は、ターゲットテーブルは以下のようになります。

| userId | name | city |__START_AT|__END_AT|
|:--|:--|:--|:--|:--|
| 123  | Isabel  | Monterrey  | 1 | 5|
| 123  | Isabel  | Chihuahua  | 5 | 6|
| 123  | Raul  | Oaxaca  | 1 | null|
| 123  | Mercedes  | Tijuana  | 2 | 5|
| 123  | Mercedes  | Mexicali  | 5 | 6|
| 123  | Mercedes  | Guadalajara  | 6 | null|
| 123  | Lily  | Cancun  | 2 | null|

## テストデータの生成

このサンプルのテストレコードを生成するには、以下の手順を踏みます。

1. Databricksワークスペースに移動し、**Create a notebook**を選択するか、サイドバーの![](https://docs.databricks.com/_images/create-icon.png)**Create**をクリックし、メニューから**Notebook**を選択します。**Create Notebook**ダイアログが表示されます。
1. **Create Notebook**ダイアログでは、**Generate test CDC records**のようなノートブック名を入力します。**Default Language**ドロップダウンメニューから**SQL**を選択します。
1. 稼働中のクラスターがある場合には、**Cluster**ドロップダウンが表示されます。ノートブックをアタッチしたいクラスターを選択します。ノートブックを作成した後に新規クラスターを[作成](https://qiita.com/taka_yayoi/items/d36a469a1e0c0cebaf1b)しアタッチすることができます。
1. **Create**をクリックします。
1. ノートブックの最初のセルに以下のクエリーを貼り付けます。

    ```sql:SQL
    CREATE SCHEMA IF NOT EXISTS cdc_data;
    
    CREATE TABLE
      cdc_data.users
    AS SELECT
      col1 AS userId,
      col2 AS name,
      col3 AS city,
      col4 AS operation,
      col5 AS sequenceNum
    FROM (
      VALUES
      -- Initial load.
      (124, "Raul",     "Oaxaca",      "INSERT", 1),
      (123, "Isabel",   "Monterrey",   "INSERT", 1),
      -- New users.
      (125, "Mercedes", "Tijuana",     "INSERT", 2),
      (126, "Lily",     "Cancun",      "INSERT", 2),
      -- Isabel is removed from the system and Mercedes moved to Guadalajara.
      (123, null,       null,          "DELETE", 6),
      (125, "Mercedes", "Guadalajara", "UPDATE", 6),
      -- This batch of updates arrived out of order. The above batch at sequenceNum 5 will be the final state.
      (125, "Mercedes", "Mexicali",    "UPDATE", 5),
      (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
      -- Uncomment to test TRUNCATE.
      -- ,(null, null,      null,          "TRUNCATE", 3)
    );
    ```

1. ノートブックを実行し、レコードを生成するには、右端にあるセルのアクションメニュー![](https://docs.databricks.com/_images/cell-actions.png)から![](https://docs.databricks.com/_images/run-cell.png)をクリックし、**Run Cell**を選択するか、**shift+enter**を押下します。

## SCDタイプ1のサンプルパイプラインを作成し実行する

1. Databricksワークスペースに移動し、**Create a notebook**を選択するか、サイドバーの![](https://docs.databricks.com/_images/create-icon.png)**Create**をクリックし、メニューから**Notebook**を選択します。**Create Notebook**ダイアログが表示されます。
1. **Create Notebook**ダイアログでは、**DLT CDC example**のようなノートブック名を入力します。使う言語に基づいて、**Default Language**ドロップダウンメニューから**Python**か**SQL**を選択します。**Cluster**はデフォルト値のままでかまいません。Delta Live Tablesランタイムはパイプラインを実行する前にクラスターを作成します。
1. **Create**をクリックします。
1. ノートブックの最初のセルにPythonかSQLクエリーを貼り付けます。
1. [パイプラインを作成](https://qiita.com/taka_yayoi/items/e4f877a64c3ae8b8c2c8#%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%81%AE%E4%BD%9C%E6%88%90)し、**Notebook Libraries**フィールドにノートブックを追加します。パイプライン処理の出力を公開するには、オプションとして**Target**フィールドにデータベース名を入力します。
1. [パイプラインをスタート](https://qiita.com/taka_yayoi/items/e4f877a64c3ae8b8c2c8#%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%E3%81%AE%E3%82%B9%E3%82%BF%E3%83%BC%E3%83%88)します。**Target**を設定した場合は、[クエリーの結果を参照し確認](https://docs.databricks.com/data/data-tab.html#view-databases-and-tables)することができます。

## サンプルクエリー

```py:Python
import dlt
from pyspark.sql.functions import col, expr

@dlt.view
def users():
  return spark.readStream.format("delta").table("cdc_data.users")

dlt.create_streaming_live_table("target")

dlt.apply_changes(
  target = "target",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  apply_as_truncates = expr("operation = 'TRUNCATE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = 1
)
```

```sql:SQL
-- Create and populate the target table.
CREATE OR REFRESH STREAMING LIVE TABLE target;

APPLY CHANGES INTO
  live.target
FROM
  stream(cdc_data.users)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
APPLY AS TRUNCATE WHEN
  operation = "TRUNCATE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 1;
```

## SCDタイプ2のサンプルパイプラインを作成し実行する

1. Databricksワークスペースに移動し、**Create a notebook**を選択するか、サイドバーの![](https://docs.databricks.com/_images/create-icon.png)**Create**をクリックし、メニューから**Notebook**を選択します。**Create Notebook**ダイアログが表示されます。
1. **Create Notebook**ダイアログでは、**DLT CDC example**のようなノートブック名を入力します。使う言語に基づいて、**Default Language**ドロップダウンメニューから**Python**か**SQL**を選択します。**Cluster**はデフォルト値のままでかまいません。Delta Live Tablesランタイムはパイプラインを実行する前にクラスターを作成します。
1. **Create**をクリックします。
1. ノートブックの最初のセルにPythonかSQLクエリーを貼り付けます。
1. [パイプラインを作成](https://qiita.com/taka_yayoi/items/e4f877a64c3ae8b8c2c8#%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%81%AE%E4%BD%9C%E6%88%90)し、**Notebook Libraries**フィールドにノートブックを追加します。パイプライン処理の出力を公開するには、オプションとして**Target**フィールドにデータベース名を入力します。
1. [パイプラインをスタート](https://qiita.com/taka_yayoi/items/e4f877a64c3ae8b8c2c8#%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%E3%81%AE%E3%82%B9%E3%82%BF%E3%83%BC%E3%83%88)します。**Target**を設定した場合は、[クエリーの結果を参照し確認](https://docs.databricks.com/data/data-tab.html#view-databases-and-tables)することができます。

## サンプルクエリー

```py:Python
import dlt
from pyspark.sql.functions import col, expr

@dlt.view
def users():
  return spark.readStream.format("delta").table("cdc_data.users")

dlt.create_streaming_live_table("target")

dlt.apply_changes(
  target = "target",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)
```

```sql:SQL
-- Create and populate the target table.
CREATE OR REFRESH STREAMING LIVE TABLE target;

APPLY CHANGES INTO
  live.target
FROM
  stream(cdc_data.users)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;
```

### Databricks 無料トライアル

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