AUTO CDC API: パイプラインを使用して変更データキャプチャを簡略化する

Lakeflow パイプラインは、 AUTO CDC API と AUTO CDC FROM SNAPSHOT API を使用して変更データキャプチャ (CDC) を簡略化します。これらの API は、CDC フィードまたはデータベーススナップショットから、緩やかに変化するディメンション (SCD) タイプ 1 とタイプ 2 の計算の複雑さを自動化します。 AUTO CDC API は、2 つの時間軸にまたがる変更を記録するバイテンポラル追跡 (ベータ) もサポートしています。 SCD タイプ 1 とタイプ 2 の詳細については、「データキャプチャとスナップショットの変更」を参照してください。二時点追跡の詳細については、Bitemporal AUTO CDCを参照してください。

注

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

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

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

SCDタイプ1ターゲットの場合、 AUTO CDC FROM SNAPSHOT フローと1つ以上の AUTO CDC フローを組み合わせることも可能です。このパターンは、ソースが定期的に権威あるスナップショットを提供し、スナップショット間に低遅延のCDCフィードを提供する場合に用いてください。

どちらの API でも、SCD タイプ 1 とタイプ 2 を使用したテーブルの更新がサポートされています。

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

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

AUTO CDC では部分的な更新もサポートされています。変更レコードは列のサブセットのみを更新します。「部分的な更新プログラムの適用」を参照してください。

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

構文とその他の参照については、 AUTO CDC INTO (パイプライン)、 create_auto_cdc_flow、およびcreate_auto_cdc_from_snapshot_flowに関するページを参照してください。

注

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

Requirements

CDC API を使用するには、サーバーレスの Lakeflow パイプラインまたは Lakeflow パイプラインPro または Advancededitions を使用するようにパイプラインを構成する必要があります。

AUTO CDC のしくみ

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

変更フィードを含むソースからの初期ハイドレートの場合は、AUTO CDC フローでonceを使用し、変更フィードの処理を続行します。 AUTO CDC を使用した外部 RDBMS テーブルのレプリケートに関するページを参照してください。

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

スナップショットからの自動 CDC のしくみ

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

AUTO CDC FROM SNAPSHOTを使用してCDC処理を行うには、ストリーミングテーブルを作成し、次にフローを定義します。

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

2 つのインジェストパターンの詳細と、それぞれを使用するタイミングについては、「スナップショット処理パターン」を参照してください。 AUTO CDC FROM SNAPSHOT の例を参照してください。

AUTO CDC FROM SNAPSHOT と AUTO CDC を組み合わせる

スナップショットバージョンとCDCシーケンスカラムは一つの順序ドメインを形成します。最新のスナップショットより新しいCDCイベントは、最新のスナップショットより優先されます。新しいスナップショットは古いCDCイベントよりも優先され、スナップショットに欠落しているキーはスナップショットバージョンで削除されたものとして扱います。新しいCDCのイベントでは、これらのキーを保存または復元することができます。

統一ターゲットは以下の要件を満たす必要があります:

ターゲットはSCDタイプ1を使用しています。
ターゲットは正確に1つの AUTO CDC FROM SNAPSHOT 流と1つ以上の AUTO CDC 流を持ちます。
各 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 では AUTO CDC FROM SNAPSHOT を、SQL では once=True を ONCE フローで使用し、AUTO CDC フローがその後の更新で新しいイベントを処理し続ける一方で、スナップショットを一度だけバックフィルします。ターゲットのフルリフレッシュはワンタイムスナップショットフローを再実行します。完全な例については、「 AUTO CDC SCD Type 1テーブルへのバックフィル追加」を参照してください。

このパターンはSCDタイプ2やバイテンポラルターゲットをサポートしません。

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

複数の列 (タイムスタンプと ID を使用してタイを解除する場合など) を順番に並べるには、 STRUCT を使用してそれらを組み合わせます。 API は最初のフィールドを優先して並べ替えを行い、値が同じ場合は 2 番目のフィールドを基準にし、その後も同様に順次判断します。

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

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

AUTO CDC の例

次の例は、変更データフィードソースを使用した SCD タイプ 1 およびタイプ 2 の処理を示しています。サンプルデータは、新しいユーザーレコードを作成し、ユーザーレコードを削除して、ユーザーレコードを更新します。 SCD タイプ 1 の例では、最後の UPDATE 操作は到着が遅れ、ターゲットテーブルから削除され、順不同のイベント処理が示されています。

これらの例で使用される入力レコードを次に示します。このデータは、「サンプルデータの作成」セクションでクエリを実行することによって作成されます。

userId	名前	city	操作	シーケンス番号
124	ラウル	Oaxaca	INSERT	1
123	Isabel	モンテレー	INSERT	1
125	メルセデス	ティファナ	INSERT	2
126	リリー	カンクン	INSERT	2
123	null 値	null 値	DELETE	6
125	メルセデス	Guadalajara	UPDATE	6
125	メルセデス	メヒカリ	UPDATE	5
123	Isabel	Chihuahua	UPDATE	5

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

userId	名前	city	操作	シーケンス番号
null 値	null 値	null 値	切り捨てる	3

注

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

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

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

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 タイプ 1 では、各レコードの最新バージョンのみが保持されます。次の例では、上記で作成した変更データフィードから読み取り、変更をストリーミングテーブルターゲットに適用します。このコードを動かすにはパイプラインが必要です。 Lakeflow パイプラインエディターを使用した ETL パイプラインの開発とデバッグを参照してください。

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

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	名前	city
124	ラウル	Oaxaca
125	メルセデス	Guadalajara
126	リリー	カンクン

ユーザー 123 (Isabel) が削除され、表示されません。 SCD Type 1 は以前の値を上書きするため、ユーザー 125 (メルセデス) には最新の市区町村 (グアダラハラ) のみが表示されます。 UPDATE の以前の sequenceNum=5 は、後に sequenceNum=6 で更新プログラムが利用可能になるため、削除されました。

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

userId	名前	city
125	メルセデス	Guadalajara

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

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

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

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 Type 2 の例を実行した後、ターゲットテーブルには次のレコードが含まれます。

userId	名前	city	__START_AT	__END_AT
123	Isabel	モンテレー	1	5
123	Isabel	Chihuahua	5	6
124	ラウル	Oaxaca	1	null 値
125	メルセデス	ティファナ	2	5
125	メルセデス	メヒカリ	5	6
125	メルセデス	Guadalajara	6	null 値
126	リリー	カンクン	2	null 値

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

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

既定では、SCD Type 2 では、列の値が変更されるたびに新しいバージョンが作成されます。追跡する列のサブセットを指定すると、新しい履歴レコードを生成するのではなく、他の列に対する変更によって現在のバージョンが更新されます。

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

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

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	名前	city	__START_AT	__END_AT
123	Isabel	Chihuahua	1	6
124	ラウル	Oaxaca	1	null 値
125	メルセデス	Guadalajara	2	null 値
126	リリー	カンクン	2	null 値

スナップショットによるAUTO CDCの例

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

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

例: 定期スナップショットを処理する

スナップショットが定期的に順番に届く場合はこの方法を活用してください。 Pythonインターフェースはバージョン管理のためにパイプラインの更新順序を使用します。 SQLの場合、例では新しいスナップショットが利用可能になったタイミングを識別するために別のバージョンテーブルを使用します。

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

手順 1: サンプルデータを作成する

スナップショットデータを含むテーブルを作成します。パイプラインの explorations フォルダーにあるノートブックまたは Databricks 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 パイプラインエディターを使用した ETL パイプラインの開発とデバッグを参照してください。

Pythonの場合、スナップショットビューのソースタイプを選択してください(サンプル作成コードでDeltaテーブルが生成されます):

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

from pyspark import pipelines as dp

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

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

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 から読み取る (クラシックコンピューティングのみ)

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

dp.create_streaming_table("target")

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

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	__END_AT
1	Oaxaca	null 値
2	モンテレー	null 値
3	ティファナ	null 値

注

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

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

ソーステーブルを更新して新しいスナップショットの到着をシミュレートします(このコードはパイプラインの explorations フォルダ内のノートブックまたは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	モンテレー	0	1
2	カーメル	1	null 値
3	ティファナ	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	キングスキャニオン

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

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

複数のスナップショットがストレージにある場合、例セレクタは利用可能なスナップショットをバージョン順に処理します。
スナップショットが順不同で到着する場合（たとえば、snapshot_3 が snapshot_4 の後に到着する場合）、サンプルセレクターは、そのバージョンが最後にコミットされたバージョンより大きくないため、それを除外します。
新しいスナップショットがなければ、バージョン関数やクエリは結果を返さず、処理も行われません。

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

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

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

snapshot_1.csv:

userId	city
1	Oaxaca
2	モンテレー
3	ティファナ

snapshot_2.csv:

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

手順 2: バージョン関数を使用して AUTO CDC FROM SNAPSHOT を実行する

パイプラインで、transformationsフォルダーに新しい Python または SQL ファイルを作成し、以下のコードを追加してください。続いて、パイプラインを実行します。 Lakeflow パイプラインエディターを使用した ETL パイプラインの開発とデバッグを参照してください。

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

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	モンテレー	1	null 値
3	ティファナ	1	null 値

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

userId	city	__START_AT	__END_AT
1	Oaxaca	1	2
2	モンテレー	1	2
2	カーメル	2	null 値
3	ティファナ	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 パイプラインを構築する

フィードバック

このページはお役に立ちましたか?

最終更新: 2026-10-09

AUTO CDC API: パイプラインを使用して変更データ キャプチャを簡略化する

Requirements

AUTO CDC のしくみ

スナップショットからの自動 CDC のしくみ

AUTO CDC FROM SNAPSHOT と AUTO CDC を組み合わせる

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

SQL

Python

AUTO CDC の例

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

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

Python

SQL

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

Python

SQL

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

Python

SQL

スナップショットによるAUTO CDCの例

例: 定期スナップショットを処理する

手順 1: サンプル データを作成する

手順 2: スナップショットから AUTO CDC を実行する

Python

SQL

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

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

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

手順 2: バージョン関数を使用して AUTO CDC FROM SNAPSHOT を実行する

Python

SQL

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

制限事項

その他のリソース

フィードバック

その他のリソース

AUTO CDC API: パイプラインを使用して変更データキャプチャを簡略化する

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

手順 1: サンプルデータを作成する

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