AUTO CDC APIs : パイプラインを使用して変更データ キャプチャを簡素化

Lakeflow pipelinesは、AUTO CDCとAUTO CDC FROM SNAPSHOTのAPIsでチェンジデータキャプチャ (CDC) を簡素化します。これらのAPIsは、CDCフィードまたはデータベースのslowly changing dimensions (SCD) タイプ1とタイプ2のいずれかから、スナップショットを計算する複雑さを自動化します。AUTO CDC APIは、2つの時間ディメンションにわたる変更を記録するバイテンポラル追跡もサポートしています（ベータ）。SCDタイプ1とタイプ2の詳細については、チェンジデータキャプチャとスナップショットを参照してください。バイテンポラル トラッキングの詳細については、バイテンポラル AUTO CDCを参照してください。

注記

AUTO CDC APIs APPLY CHANGES APIsに代わるもので、同じ構文を持ちます。 APPLY CHANGES APIs引き続き使用できますが、 Databricks代わりにAUTO CDC APIs使用することをお勧めします。

使用する API は、変更データのソースによって異なります。

• AUTO CDC : ソース データベースで CDC フィードが有効になっている場合に使用します。AUTO CDCは変更データフィード (CDF) からの変更を処理します。 パイプライン SQL と Python インターフェースの両方でサポートされています。
• AUTO CDC FROM SNAPSHOT ：ソースデータベースで CDC が有効になっておらず、スナップショットのみが利用可能な場合に使用します。この API はスナップショットを比較して変更を特定し、それらを処理します。パイプラインの SQL インターフェイスと Python インターフェイスの両方でサポートされています。

SCD タイプ 1 のターゲットの場合、1 つ以上の AUTO CDC フローと AUTO CDC FROM SNAPSHOT フローを組み合わせることもできます。ソースから定期的信頼できるスナップショットと、スナップショット間の低レイテンシーのCDCフィードが提供される場合は、このパターンを使用します。

どちらのAPIs SCDタイプ 1 とタイプ 2 を使用したテーブルの更新をサポートしています。

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

AUTO CDC のみに、SCDタイプ2の履歴を拡張し、ビジネス時間とシステム時間という2つの時間ディメンションにわたる変更を追跡するバイテンポラルストレージも使用できます。バイテンポラルはベータ版です。バイテンポラル AUTO CDCを参照してください。

AUTO CDC 部分的な更新もサポートしており、変更レコードは列のサブセットのみを更新します。部分的な更新を適用を参照してください。

AUTO CDC APIs 、 Apache Spark 宣言型パイプラインではサポートされていません。

構文およびその他のリファレンスについては、 「AUTO CDC INTO (パイプライン)」 、 「create_auto_cdc_flow」 、および「create_auto_cdc_from_snapshot_flow 」を参照してください。

注記

これらの APIsを使用して、ソースデータの変更に基づき、パイプラインのテーブルを更新します。Delta テーブルの行レベルの変更情報を記録してクエリーする方法については、「Databricksでチェンジデータフィードを使用する」を参照してください。

要件

CDC APIsを使用するには、パイプラインは、LakeFlow Pipelines Pro または Advanced エディションを使用するように構成する必要があります。

AUTO CDCの仕組み

AUTO CDCでCDC処理を実行するには、ストリーミング テーブルを作成し、 SQLのAUTO CDC ... INTOステートメントまたはPythonのcreate_auto_cdc_flow()関数を使用して、変更フィードのソース、キー、シーケンスを指定します。 シーケンスとSCDロジックの仕組みの説明については、 「チェンジデータ キャプチャ」と「スナップショット」を参照してください。 AUTO CDC の例を参照してください。

変更フィードを含むソースからの初期ハイドレーションの場合は、 onceフローでAUTO CDC使用し、変更フィードの処理を続行します。「AUTO CDC を使用して外部 RDBMS テーブルを複製する」を参照してください。

構文の詳細については、 「AUTO CDC INTO (パイプライン)」または「create_auto_cdc_flow」を参照してください。

スナップショットからの自動CDC仕組み

AUTO CDC FROM SNAPSHOT 順序付きのスナップショットを比較することで、ソースデータの変更を決定します。Delta テーブル、クラウドストレージファイル、または JDBC から直接スナップショットを読み取ることができます。パイプラインの SQL および Python インターフェイスの両方でサポートされています。

AUTO CDC FROM SNAPSHOTでCDC処理を実行するには、ストリーミングテーブルを作成し、フローを定義します。

• SQL では、CREATE STREAMING TABLE の CREATE FLOW ステートメントの AUTO CDC ... FROM SNAPSHOT 形式または埋め込み FLOW AUTO CDC 句を使用します。ソースは必須の FROM SNAPSHOT (snapshot_query) 句と、処理する次のスナップショットバージョンを選択するオプションの WITH VERSION (version_query) 句です。CREATE FLOW (パイプライン) を参照してください。
• Pythonでは、create_auto_cdc_from_snapshot_flow()関数を使用してスナップショット、キー、およびその他の引数を指定します。create_auto_cdc_from_snapshot_flow を参照してください。

2 つの取り込みパターンとそれぞれの使用時期の詳細については、スナップショット処理パターンを参照してください。AUTO CDC FROM スナップショット の例を参照してください。

AUTO CDC FROM スナップショット と AUTO CDC の組み合わせ

スナップショットのバージョンとCDCのシーケンス列によって、1つの順序付けドメインが形成されます。最新のスナップショットよりも新しいCDCイベントは、スナップショットよりも優先されます。新しいスナップショットは、古いCDCイベントよりも優先され、スナップショットに存在しないキーは、スナップショットのバージョンで削除されたものとして扱われます。新しいCDCイベントによって、それらのキーを保持または復元できます。

統合ターゲットは、次の要件を満たしている必要があります：

• ターゲットは SCD タイプ 1 を使用します。
• ターゲットには、AUTO CDC FROM SNAPSHOT フローが正確に 1 つと、AUTO CDC フローが 1 つ以上あります。
• 各 AUTO CDC フローには一意の名前があります。
• すべてのフローは、同じ順序で同数のキーを使用します。スナップショットフローのキー名は、AUTO CDC のキー名の大文字と小文字を区別せずに比較されます。複数の AUTO CDC フローでは、同一のキー名と大文字・小文字を使用する必要があります。
• スナップショットバージョンと各 CDC シーケンス列のデータ型は完全に同じです。
• AUTO CDC FROM SNAPSHOT フローではエクスペクテーションが定義されていません。
• AUTO CDC フローでは、IGNORE NULL UPDATES またはその列リストのバリエーションは使用されません。
• パイプラインはトリガーモードを使用します。
• 既存の AUTO CDC FROM SNAPSHOT ターゲットに AUTO CDC フローを追加する場合、ターゲットには、予約済みの AUTO CDC システム列と名前が競合するユーザー列が含まれていてはなりません。

AUTO CDC フローと AUTO CDC FROM SNAPSHOT フローの両方で、Python または SQL のパイプライン インターフェイスを使用できます。同じターゲット内で SQL フローと Python フローを混在させることができます。

Python の once=True または SQL の ONCE を AUTO CDC FROM SNAPSHOT フローで使用してスナップショットを 1 回バックフィルし、以降の更新では AUTO CDC フローが引き続き新しいイベントを処理できるようにします。ターゲットの完全更新により、1回限りのスナップショットフローが再実行されます。完全な例については、AUTO CDC SCD Type 1 テーブルへのバックフィルの追加を参照してください。

このパターンでは、SCD Type 2 またはバイテンポラル ターゲットはサポートされていません。

シーケンスには複数の列を使用する

複数の列で順序付けるには（たとえば、同点を区切るためのタイムスタンプと ID）、 STRUCTを使用してそれらを結合します。API はまず最初のフィールドで順序付けし、同点の場合は 2 番目のフィールドを考慮します。

SQL

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

Python

sequence_by = struct("timestamp_col", "id_col")

AUTO CDCの例

次の例は、チェンジデータフィード ソースを使用したSCDタイプ 1 およびタイプ 2 の処理を示しています。 サンプル データは、新しいユーザー レコードを作成し、ユーザー レコードを削除し、ユーザー レコードを更新します。SCD タイプ 1 の例では、最後の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

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

サンプル データ生成クエリの最後の行のコメントを解除すると、 sequenceNum=3でテーブルを切り捨てる (テーブルをクリアする) ことを指定する次のレコードが挿入されます。

userId	name	city	operation	sequenceNum
null	null	null	TRUNCATE	3

userId	name	city	operation	sequenceNum
null	null	null	TRUNCATE	3

注記

以下のすべての例には、 DELETEとTRUNCATE両方の操作を指定するオプションが含まれていますが、それぞれはオプションです。

サンプルデータを作成する

次のステートメントを実行してサンプル データセットを作成します。このコードは、パイプライン定義の一部として実行されることを意図したものではありません。変換フォルダーではなく、パイプラインの探索フォルダーから実行します。

SQL

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.users_cdf
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 batch at sequenceNum 6 is the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

SCDタイプ1の更新を処理する

SCD Type 1は、各レコードの最新バージョンのみを保持します。次の例では、上記で作成したチェンジデータフィードから読み取りを行い、ストリーミングテーブルのターゲットに変更を適用します。このコードを実行するにはパイプラインが必要です。LakeFlow Pipelines Editor を使用したETLパイプラインの開発とデバッグを参照してください。

Python

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_current")

dp.create_auto_cdc_flow(
  target = "users_current",
  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 OR REFRESH STREAMING TABLE users_current;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_current
FROM
  stream(main.cdc_tutorial.users_cdf)
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タイプ1の例を実行すると、ターゲットテーブルには次のレコードが含まれます：

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

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

ユーザー 123 (Isabel) は削除されたため、表示されません。ユーザー 125 (メルセデス) には、SCD タイプ 1 が以前の値を上書きするため、最新の都市 (グアダラハラ) のみが表示されます。sequenceNum=5の以前のUPDATEは、 sequenceNum=6の後続の更新が到着したため削除されました。

TRUNCATEレコードのコメントを解除して例を実行すると、テーブルはsequenceNum=3でクリアされます。これは、レコード124と126テーブル内に存在せず、最終的なターゲット テーブルには次のレコードのみが含まれていることを意味します。

userId	name	city
125	Mercedes	Guadalajara

userId	name	city
125	Mercedes	Guadalajara

SCDタイプ2の更新を処理する

SCD タイプ 2 は、レコードの各バージョンに対して新しい行を作成し、各バージョンがアクティブだった時期を__START_AT列と__END_AT列で示すことで、変更の完全な履歴を保存します。

Python

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  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 OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;

SCDタイプ2の例を実行した後、ターゲットテーブルには以下のレコードが含まれます：

userId	name	city	__START_AT	__END_AT
123	Isabel	Monterrey	1	5
123	Isabel	Chihuahua	5	6
124	Raul	Oaxaca	1	null
125	Mercedes	Tijuana	2	5
125	Mercedes	Mexicali	5	6
125	Mercedes	Guadalajara	6	null
126	Lily	Cancun	2	null

userId	name	city	__START_AT	__END_AT
123	Isabel	Monterrey	1	5
123	Isabel	Chihuahua	5	6
124	Raul	Oaxaca	1	null
125	Mercedes	Tijuana	2	5
125	Mercedes	Mexicali	5	6
125	Mercedes	Guadalajara	6	null
126	Lily	Cancun	2	null

テーブルには完全な履歴が保存されます。ユーザー 123 には 2 つのバージョンがあります (削除時にシーケンス 6 で終了)。ユーザー 125 には、都市の変更を示す 3 つのバージョンがあります。__END_AT = nullのレコードは現在アクティブです。

SCD タイプ 2 で列のサブセットを追跡する

デフォルトでは、SCD タイプ 2 は列の値が変更されるたびに新しいバージョンを作成します。追跡する列のサブセットを指定して、他の列を変更した場合に新しい履歴レコードを生成するのではなく、現在のバージョンをその場で更新することができます。

次の例では、 city列を履歴追跡から除外します。

Python

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2",
  track_history_except_column_list = ["city"]
)

SQL

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2
TRACK HISTORY ON * EXCEPT
  (city)

city変更は追跡されないため、都市の更新では新しいバージョンが作成されるのではなく、現在の行が上書きされます。ターゲット テーブルには次のレコードが含まれています。

userId	name	city	__START_AT	__END_AT
123	Isabel	Chihuahua	1	6
124	Raul	Oaxaca	1	null
125	Mercedes	Guadalajara	2	null
126	Lily	Cancun	2	null

userId	name	city	__START_AT	__END_AT
123	Isabel	Chihuahua	1	6
124	Raul	Oaxaca	1	null
125	Mercedes	Guadalajara	2	null
126	Lily	Cancun	2	null

スナップショットからの自動CDCの例

次のセクションでは、 AUTO CDC FROM SNAPSHOTを使用してスナップショットを SCD タイプ 1 またはタイプ 2 のターゲット テーブルに処理する例を示します。このAPIを使用する場合の背景については、 「チェンジデータ キャプチャ」と「スナップショット」を参照してください。

このセクションの例には、Python および SQL の実装が含まれています。完全な SQL 構文については、CREATE FLOW (パイプライン) を参照してください。

例: 定期的なスナップショットの処理

スナップショットが定期的に、かつ順序どおりに到着する場合は、このアプローチを使用します。Python インターフェイスでは、バージョニングにパイプラインの更新順序が使用されます。SQL の場合、この例では別のバージョンテーブルを使用して、新しいスナップショットがいつ利用可能になるかを識別します。

Delta テーブル、クラウド ストレージ ファイル、JDBC 接続など、複数のソース タイプからスナップショットを読み取ることができます。

ステップ1: サンプルデータを作成する

スナップショット データを含むテーブルを作成します。パイプラインのexplorationsフォルダーにあるノートブックまたは Databricks SQL から次のコードを実行します。

SQL

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.snapshot (
  userId INT,
  city STRING
);

CREATE TABLE main.cdc_tutorial.snapshot_version (
  version BIGINT
);

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (1, 'Oaxaca'),
  (2, 'Monterrey'),
  (3, 'Tijuana');

INSERT INTO main.cdc_tutorial.snapshot_version VALUES (0);

ステップ 2: スナップショットから AUTO CDCを実行

このステップのコードを実行するには、パイプラインが必要です。詳細については、Lakeflow Pipelines EditorによるETLパイプラインの開発とデバッグを参照してください。

Pythonの場合は、スナップショットビューのソースタイプを選択します（サンプル作成コードによってDeltaテーブルが生成されます）。

オプションA: Deltaテーブルから読み取る

Python

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")

オプションB: クラウドストレージから読み取る

Python

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")

オプション C: JDBCからの読み取り (クラシック コンピュートのみ)

Python

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return (spark.read
    .format("jdbc")
    .option("url", "<jdbc-url>")
    .option("dbtable", "<table-name>")
    .option("user", "<username>")
    .option("password", "<password>")
    .load()
  )

次に、ターゲットテーブルとフローを追加します。SQL の実装では、snapshot_version テーブルを使用して新しいスナップショットがいつ利用可能になるかを特定します。

Python

Python

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = "source",
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
FROM SNAPSHOT (
  SELECT userId, city
  FROM main.cdc_tutorial.snapshot
)
WITH VERSION (
  SELECT version
  FROM main.cdc_tutorial.snapshot_version
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
)
KEYS (userId)
STORED AS SCD TYPE 2;

最初のパイプラインの実行後、すべてのレコードがアクティブ行として挿入されます。

userId	city	__START_AT	__END_AT
1	Oaxaca	0	null
2	Monterrey	0	null
3	Tijuana	0	null

userId	city	__START_AT	__END_AT
1	Oaxaca	0	null
2	Monterrey	0	null
3	Tijuana	0	null

注記

代わりにSCDタイプ1を使用し、現在の状態のみを保持するには、Pythonでstored_as_scd_type=1を設定するか、SQLでSTORED AS SCD TYPE 1を設定します。この場合、ターゲットテーブルには__START_ATおよび__END_ATの列が含まれていません。

ステップ 3: 新しいスナップショットをシミュレートして再実行します

ソーステーブルを更新して、新しいスナップショットの到着をシミュレートします（パイプラインの explorations フォルダにあるノートブックまたはSQLファイルからこのコードを実行します）。

SQL

TRUNCATE TABLE main.cdc_tutorial.snapshot;

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley'),
  (6, 'Kings Canyon');

UPDATE main.cdc_tutorial.snapshot_version SET version = 1;

パイプラインを再度実行します。AUTO CDC FROM SNAPSHOT新しいスナップショットを以前のスナップショットと比較し、ユーザー 1 が削除され、ユーザー 2 と 3 が更新され、ユーザー 4 と 6 が挿入されたことを検出します。これにより変更フィードが生成され、 AUTO CDCを使用して出力テーブルが作成されます。

SCD タイプ 2 で 2 回目の実行を行うと、ターゲット テーブルに次のレコードが含まれます。

userId	city	__START_AT	__END_AT
1	Oaxaca	0	1
2	Monterrey	0	1
2	カーメル	1	null
3	Tijuana	0	1
3	ロサンゼルス	1	null
4	デスバレー	1	null
6	キングスキャニオン	1	null

userId	city	__START_AT	__END_AT
1	Oaxaca	0	1
2	Monterrey	0	1
2	カーメル	1	null
3	Tijuana	0	1
3	ロサンゼルス	1	null
4	デスバレー	1	null
6	キングスキャニオン	1	null

ユーザー1は終了（削除）されました。ユーザー 2 と 3 には、それぞれ都市の変更を示す 2 つのバージョンがあります。ユーザー4と6が新たに挿入されました。

SCD タイプ 1 で 2 回目の実行を行うと、ターゲット テーブルには現在の状態のみが表示されます。

userId	city
2	カーメル
3	ロサンゼルス
4	デスバレー
6	キングスキャニオン

userId	city
2	カーメル
3	ロサンゼルス
4	デスバレー
6	キングスキャニオン

例: バージョン関数を使用してスナップショットを処理する

スナップショットの順序を明示的に制御する必要がある場合は、このアプローチを使用します。たとえば、複数のスナップショットが同時に到着する場合や、スナップショットが順序通りに到着しない場合に、このアプローチを使用します。次のスナップショットとそのバージョン番号を選択する方法を定義します。APIは、バージョンの昇順でスナップショットを処理します。

• ストレージに複数のスナップショットが存在する場合、例セレクターは利用可能なスナップショットをバージョン順に処理します。
• スナップショットが順序どおりに到着しなかった場合（例：snapshot_3 が snapshot_4 の後に到着した場合）、バージョンが最後にコミットされたバージョンより大きくないため、例セレクターによって除外されます。
• 新しいスナップショットがない場合、バージョン関数またはクエリーは結果を返さず、処理は行われません。

ステップ 1: スナップショット ファイルを準備する

スナップショット データを含む CSV ファイルを作成し、ボリュームまたはクラウド ストレージの場所に追加します。ファイルに時系列順に名前を付けます (例: snapshot_1.csv 、 snapshot_2.csv )。

各ファイルには、 userIdとcity列が含まれている必要があります。例えば：

snapshot_1.csv :

userId	city
1	Oaxaca
2	Monterrey
3	Tijuana

userId	city
1	Oaxaca
2	Monterrey
3	Tijuana

snapshot_2.csv :

userId	city
2	カーメル
3	ロサンゼルス
4	デスバレー

userId	city
2	カーメル
3	ロサンゼルス
4	デスバレー

ステップ 2: バージョン機能を使用して AUTO CDC FROM スナップショットを実行する

パイプラインで、transformations フォルダーに新しい Python または SQL ファイルを作成し、次のコードを追加します。その後、パイプラインをランします。Lakeflow Pipelines Editor を使用した ETL パイプラインの開発とデバッグを参照してください。

Python

Python

from pyspark import pipelines as dp
from typing import Optional, Tuple
from pyspark.sql import DataFrame

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Optional[Tuple[DataFrame, int]]:
  snapshot_dir = "/Volumes/main/cdc_tutorial/snapshots/" # or the location you created the sample data

  files = dbutils.fs.ls(snapshot_dir)
  snapshot_files = [f.name for f in files if f.name.startswith("snapshot_") and f.name.endswith(".csv")]

  snapshot_versions = []
  for filename in snapshot_files:
    try:
      version = int(filename.replace("snapshot_", "").replace(".csv", ""))
      snapshot_versions.append(version)
    except ValueError:
      continue

  snapshot_versions.sort()

  if latest_snapshot_version is None:
    if snapshot_versions:
      next_version = snapshot_versions[0]
    else:
      return None
  else:
    next_versions = [v for v in snapshot_versions if v > latest_snapshot_version]
    if next_versions:
      next_version = next_versions[0]
    else:
      return None

  snapshot_path = f"{snapshot_dir}snapshot_{next_version}.csv"
  df = spark.read.format("csv").option("header", True).load(snapshot_path)
  return (df, next_version)


dp.create_streaming_table("main.cdc_tutorial.target_versioned")

dp.create_auto_cdc_from_snapshot_flow(
  target = "main.cdc_tutorial.target_versioned",
  source = next_snapshot_and_version,
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

SQL

CREATE OR REFRESH STREAMING TABLE main.cdc_tutorial.target_versioned;

CREATE FLOW target_versioned_snapshot_flow AS
AUTO CDC INTO main.cdc_tutorial.target_versioned
FROM SNAPSHOT (
  SELECT CAST(userId AS INT) AS userId, city
  FROM read_files(
    '/Volumes/main/cdc_tutorial/snapshots/',
    format => 'csv',
    header => true
  )
  WHERE _metadata.file_name = CONCAT(
    'snapshot_',
    CAST((SELECT version FROM current_snapshot_version()) AS STRING),
    '.csv'
  )
)
WITH VERSION (
  WITH snapshot_files AS (
    SELECT
      CAST(REGEXP_EXTRACT(path, 'snapshot_([0-9]+)[.]csv$', 1) AS BIGINT) AS version
    FROM list_files('/Volumes/main/cdc_tutorial/snapshots/')
    WHERE path RLIKE 'snapshot_[0-9]+[.]csv$'
  )
  SELECT version
  FROM snapshot_files
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY version
  LIMIT 1
)
KEYS (userId)
STORED AS SCD TYPE 2;

注記

代わりにSCDタイプ1を使用するには、Pythonでstored_as_scd_type=1を設定するか、SQLでSTORED AS SCD TYPE 1を設定します。

snapshot_1.csvを処理した後、ターゲット テーブルには次のレコードが含まれます。

userId	city	__START_AT	__END_AT
1	Oaxaca	1	null
2	Monterrey	1	null
3	Tijuana	1	null

userId	city	__START_AT	__END_AT
1	Oaxaca	1	null
2	Monterrey	1	null
3	Tijuana	1	null

snapshot_2.csvを処理した後、ターゲット テーブルには次のレコードが含まれます。

userId	city	__START_AT	__END_AT
1	Oaxaca	1	2
2	Monterrey	1	2
2	カーメル	2	null
3	Tijuana	1	2
3	ロサンゼルス	2	null
4	デスバレー	2	null

userId	city	__START_AT	__END_AT
1	Oaxaca	1	2
2	Monterrey	1	2
2	カーメル	2	null
3	Tijuana	1	2
3	ロサンゼルス	2	null
4	デスバレー	2	null

注記

SCD タイプ 1 の場合、テーブルは最新のスナップショットとまったく同じになることに注意してください。違いは、ダウンストリーム クエリでは変更フィードを使用して変更されたレコードのみを処理できることです。

ステップ 3: 新しいスナップショットを追加する

変更されたデータ (都市の値の変更、新しい行、削除された行など) を含む新しい CSV ファイルを保存場所に追加します。次に、パイプラインを再度実行して、新しいスナップショットを処理します。

制限事項

• シーケンス列は並べ替え可能なデータ型である必要があります。NULLシーケンス値はサポートされていません。
• AUTO CDCプロセスのターゲットからデータをストリームするには、その変更フィードから読み取ります。 詳細については、 「AUTO CDCターゲット テーブルから変更データフィードを読み取る」を参照してください。

その他のリソース

• 変更データ キャプチャとスナップショット: CDC概念、スナップショット、 SCDタイプについて学びます。
• AUTO CDCを使用して外部 RDBMS テーブルを複製します。 onceフローを使用して初期ハイドレーションを実行し、変更の処理を続行する方法を学習します。
• パイプラインを使用したヒストリカルデータのバックフィル: 追加専用および AUTO CDC SCD Type 1 ターゲットをバックフィルする方法について説明します。
• 高度な AUTO CDCトピック: AUTO CDCターゲットでの変更操作、変更データフィードの読み取り、メトリクスの処理について学びます。
• パイプラインを巻き戻して再実行：AUTO CDCターゲットを以前の時点に復元し、変更を再処理する方法をご覧ください。
• チュートリアル: チェンジデータキャプチャを使用してETLパイプラインを構築する