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 はスナップショットを比較して変更を判断し、処理します。Python インターフェースでのみサポートされます。

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

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

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

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

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

注記

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

要件

CDC APIを使用するには、パイプラインをServerless LakeFlow Pipelines、または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 順序どおりのスナップショットを比較してソース データの変更を判別します。Python パイプライン インターフェースでのみサポートされます。スナップショットは、Delta テーブル、クラウド ストレージ ファイル、または JDBC から直接読み取ることができます。

AUTO CDC FROM SNAPSHOTでCDC処理を実行するには、ストリーミング テーブルを作成し、 create_auto_cdc_from_snapshot_flow()関数を使用してスナップショット、キー、その他の引数を指定します。 2 つの取り込みパターンの詳細と、それぞれの使用タイミングについては、 「スナップショット処理パターン」を参照してください。AUTO CDC FROM スナップショットの例を参照してください。

構文の詳細については、 create_auto_cdc_from_snapshot_flow を参照してください。

バイテンポラル AUTO CDC の仕組み

備考

ベータ版

バイテンポラル AUTO CDC は ベータ版です。

SCD タイプ 1 とタイプ 2 はユニテンポラルです。これらは単一の時間ディメンションにわたる変更を追跡します。バイテンポラルは、SCD タイプ 2 の履歴を拡張し、2 つの時間ディメンションにわたる変更を追跡し、2 つの視点を区別します：

• ビジネス時間 ：イベントが実際に発生した時間です。
• システム時刻 : システムがイベントを記録または取り込んだ時刻です。

SCDタイプ2と同様に、バイテンポラルはレコードの完全な履歴を保持します。これにより、2つ目のタイムラインが追加され、過去のどの時点でもデータが示した内容とシステムが認識していた内容の両方を再構築できます。

たとえば、ヘッジファンドはソースシステムから株式データを取り込みます。Acme Corpの株価は1月1日に変動しますが、ファンドは1月5日までその更新を取り込みません。バイテンポラルAUTO CDCにより、ファンドは2つの異なる質問に回答できます。1月1日（ビジネス時間）時点でのAcme Corpの実際の株価はいくらだったか、そして、ファンドが1月3日（システム時間）に取引の決定を下したときにシステムが信じていた価格はいくらだったか。これらのタイムラインを区別する機能は、監査、規制報告、および財務上の意思決定に役立ちます。

バイテンポラル処理を有効にするには、STORED AS BITEMPORAL (SQL) または stored_as_scd_type="bitemporal" (Python) を設定し、ビジネス時間列には SEQUENCE BY を使用し、システム時間列には SYSTEM SEQUENCE BY を使用します。ターゲットテーブルは、SCD タイプ 2 の __START_AT 列と __END_AT 列に加えて、__SYSTEM_START_AT 列と __SYSTEM_END_AT 列を追加します。構文の詳細については、AUTO CDC INTO (パイプライン) または create_auto_cdc_flow を参照してください。

挿入、更新、順不同の更新、および削除がバイテンポラルテーブルにどのように影響するかについては、バイテンポラル AUTO CDC の例で詳細な手順をご覧ください。

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

複数の列で順序付けるには（たとえば、同点を区切るためのタイムスタンプと 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 タイプ 1 は各レコードの最新バージョンのみを保持します。以下の例では、上で作成されたチェンジデータフィードから読み取り、ストリーミングテーブルのターゲットに変更を適用します。パイプラインとは？このコードを実行します。

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を使用する場合の背景については、 「チェンジデータ キャプチャ」と「スナップショット」を参照してください。

例: パイプラインの取り込み時間を使用してスナップショットを処理する

スナップショットが定期的かつ順番に到着し、パイプライン実行タイムスタンプをバージョン管理に利用できる場合は、このアプローチを使用します。パイプラインが更新されるたびに、新しいスナップショットが取り込まれます。

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
);

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

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

パイプラインとは？このステップでコードを実行する

スナップショット ビューのソース タイプを選択します (サンプル作成コードは 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()
  )

すべてのオプション、ターゲットに書き込む

次に、ターゲット テーブルとフローを追加します。

Python

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = "source",
  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 を使用し、現在の状態のみを保持するには、 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');

パイプラインを再度実行します。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後に到着した場合）、そのスナップショットはスキップされます。
• 新しいスナップショットがない場合、関数はNoneを返し、処理は行われません。

ステップ 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 スナップショットを実行する

新しいノートブックを作成し、次のパイプラインコードを貼り付けます。では、パイプラインとは？。

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
)

注記

代わりに SCD タイプ 1 を使用するには、 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 ファイルを保存場所に追加します。次に、パイプラインを再度実行して、新しいスナップショットを処理します。

バイテンポラル AUTO CDC の例

備考

ベータ版

バイテンポラル AUTO CDC は ベータ版です。

次の例では、少数の合成CDCイベントからバイテンポラルなターゲットテーブルを作成します。bt列にはビジネス時間が含まれ、st列にはシステム時間が含まれます。

Python

Python

from pyspark import pipelines as dp

# Source: synthetic CDC events
dp.create_streaming_table(name="cdc_source")

@dp.append_flow(target="cdc_source", once=True)
def load_cdc_source():
  return spark.createDataFrame(
    [
      (1, "x10", "y10", 10, 100),
      (1, "x20", "y20", 20, 200)
    ],
    schema="id INT, x STRING, y STRING, bt INT, st INT",
  )

# Target: bitemporal table
dp.create_streaming_table(name="target_bitemporal")

dp.create_auto_cdc_flow(
  target = "target_bitemporal",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "bt",
  system_sequence_by = "st",
  stored_as_scd_type = "bitemporal"
)

SQL

SQL

-- Source: synthetic CDC events
CREATE OR REFRESH STREAMING TABLE cdc_source_sql;

CREATE FLOW cdc_source_sql AS INSERT INTO ONCE
  cdc_source_sql BY NAME
SELECT * FROM VALUES
  (1, 'x10', 'y10', 10, 100),
  (1, 'x20', 'y20', 20, 200)
  AS t(id, x, y, bt, st);

-- Target: bitemporal table
CREATE OR REFRESH STREAMING TABLE target_bitemporal_sql;

CREATE FLOW target_bitemporal_sql AS AUTO CDC INTO
  target_bitemporal_sql
FROM
  stream(cdc_source_sql)
KEYS
  (id)
SEQUENCE BY
  bt
SYSTEM SEQUENCE BY
  st
STORED AS
  BITEMPORAL;

以下のステップでは、バイテンポラルテーブルが単一企業の挿入、更新、順不同の更新、および削除を記録する方法について説明します。シーケンス列は__START_ATと__END_AT（ビジネス時間）列を生成し、システムシーケンス列は__SYSTEM_START_ATと__SYSTEM_END_AT（システム時間）列を生成します。

列	説明
__START_AT	この行が有効になったビジネス時間です。
__END_AT	この行の有効期間が終了するビジネス時間。null は、無期限に有効な場合。
__SYSTEM_START_AT	この行のデータとビジネス時間間隔が真であると認識されているシステム時間です。
__SYSTEM_END_AT	この行のデータとビジネス時間間隔が無効化されることが認識されているシステム時刻。null は、無期限に真であることが知られている場合。

列	説明
__START_AT	この行が有効になったビジネス時間です。
__END_AT	この行の有効期間が終了するビジネス時間。null は、無期限に有効な場合。
__SYSTEM_START_AT	この行のデータとビジネス時間間隔が真であると認識されているシステム時間です。
__SYSTEM_END_AT	この行のデータとビジネス時間間隔が無効化されることが認識されているシステム時刻。null は、無期限に真であることが知られている場合。

システムは、両方のタイムラインで任意の順序で到着するイベントを処理します。イベントが、すでに処理されたイベントよりも前のビジネス時間またはシステム時間で到着した場合、システムは末尾にのみ追加するのではなく、影響を受ける履歴を修正します。

ステップ1：挿入

会社Aは2025年7月18日 10:01:00（ビジネス時間）に追加されますが、10:05:00（システム時間）まで取り込まれません。

入力：

CompanyId	データポイント	シーケンス	システムシーケンス	オペレーション
A	XFv1	2025年7月18日 10時01分00秒	2025/07/18 10:05:00	INSERT

CompanyId	データポイント	シーケンス	システムシーケンス	オペレーション
A	XFv1	2025年7月18日 10時01分00秒	2025/07/18 10:05:00	INSERT

出力：

CompanyId	データポイント	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	2025年7月18日 10時01分00秒	NULL	2025/07/18 10:05:00	NULL

CompanyId	データポイント	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	2025年7月18日 10時01分00秒	NULL	2025/07/18 10:05:00	NULL

XFv1 は、10時01分00秒から有効で、終了時期は不明です。システムはこの事実をシステム時刻 10時05分00秒 に認識しましたが、終了時期は不明です。

ステップ 2: 更新

会社Aは2025/07/18 12:15:43（ビジネス時間）に更新され、システムは12:20:00（システム時間）にイベントを処理します。システムは、更新が既知となる前に信頼していたものと、更新が取り込まれた後の修正されたビジネス履歴の両方を保持します。

入力：

CompanyId	データポイント	シーケンス	システムシーケンス	オペレーション
A	XFv2	2025年7月18日 12時15分43秒	2025年7月18日 12:20:00	UPDATE

CompanyId	データポイント	シーケンス	システムシーケンス	オペレーション
A	XFv2	2025年7月18日 12時15分43秒	2025年7月18日 12:20:00	UPDATE

出力：

CompanyId	データポイント	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	2025年7月18日 10時01分00秒	NULL	2025/07/18 10:05:00	2025年7月18日 12:20:00
A	XFv1	2025年7月18日 10時01分00秒	2025年7月18日 12時15分43秒	2025年7月18日 12:20:00	NULL
A	XFv2	2025年7月18日 12時15分43秒	NULL	2025年7月18日 12:20:00	NULL

CompanyId	データポイント	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	2025年7月18日 10時01分00秒	NULL	2025/07/18 10:05:00	2025年7月18日 12:20:00
A	XFv1	2025年7月18日 10時01分00秒	2025年7月18日 12時15分43秒	2025年7月18日 12:20:00	NULL
A	XFv2	2025年7月18日 12時15分43秒	NULL	2025年7月18日 12:20:00	NULL

XFv1 は 10:01:00 から有効であると信じられており、終了時刻は不明でした。システムはその信念を 10:05:00 から 12:20:00 まで保持していました。XFv1 は現在、12:15:43 までのみ有効であることが知られています。これは、システム時刻 12:20:00 から有効な、終了時刻不明の修正履歴です。XFv2 は 12:15:43 から有効で、終了時刻は不明であり、システム時刻 12:20:00 に認識されました。

ステップ3: 順不同の更新

順不同の更新が到着し、それが会社Aが2025/07/18 12:05:00（営業時間）に実際に更新されたことを示していますが、12:25:00（システム時間）まで取り込まれません。システム時間で遅れて到着したが、それよりも前の営業時間を持つ更新が届いた場合、システムは過去の営業時間を修正し、順不同の更新前時点での認識と、修正後の履歴の両方を保持します。

入力：

CompanyId	データポイント	シーケンス	システムシーケンス	オペレーション
A	XFv3	2025/07/18 12:05:00	2025年7月18日 12:25:00	UPDATE

CompanyId	データポイント	シーケンス	システムシーケンス	オペレーション
A	XFv3	2025/07/18 12:05:00	2025年7月18日 12:25:00	UPDATE

出力：

CompanyId	データポイント	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	2025年7月18日 10時01分00秒	NULL	2025/07/18 10:05:00	2025年7月18日 12:20:00
A	XFv1	2025年7月18日 10時01分00秒	2025年7月18日 12時15分43秒	2025年7月18日 12:20:00	2025年7月18日 12:25:00
A	XFv1	2025年7月18日 10時01分00秒	2025/07/18 12:05:00	2025年7月18日 12:25:00	NULL
A	XFv3	2025/07/18 12:05:00	2025年7月18日 12時15分43秒	2025年7月18日 12:25:00	NULL
A	XFv2	2025年7月18日 12時15分43秒	NULL	2025年7月18日 12:20:00	NULL

CompanyId	データポイント	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	2025年7月18日 10時01分00秒	NULL	2025/07/18 10:05:00	2025年7月18日 12:20:00
A	XFv1	2025年7月18日 10時01分00秒	2025年7月18日 12時15分43秒	2025年7月18日 12:20:00	2025年7月18日 12:25:00
A	XFv1	2025年7月18日 10時01分00秒	2025/07/18 12:05:00	2025年7月18日 12:25:00	NULL
A	XFv3	2025/07/18 12:05:00	2025年7月18日 12時15分43秒	2025年7月18日 12:25:00	NULL
A	XFv2	2025年7月18日 12時15分43秒	NULL	2025年7月18日 12:20:00	NULL

XFv1 は 10:01:00 から 12:15:43 まで有効であると認識されていました。そして、その有効性はシステム時間 12:25:00 まで有効であると認識されています。新しいアップデートにより、XFv1 のビジネス有効期間が 12:05:00 に終了するように修正されます。これは、システム時間 12:25:00 から有効な修正履歴です。XFv3 は現在、12:05:00 から 12:15:43 まで有効であることが知られています。この有効性は、システム時間 12:25:00 から期限なしで有効であると認識されています。

ステップ 4: 削除

会社Aは2025年7月18日 12:30:00に削除され、システムは12:30:00にイベントを消費します。削除操作はエンティティのビジネス上の存在の終了を表すため、システムは代替行を作成しません。XFv2は2行に表示され、会社が存在しなくなったときとシステムが削除を認識したときの両方の完全な監査証跡を保持します。

入力：

CompanyId	データポイント	シーケンス	システムシーケンス	オペレーション
A	XFv2	2025/07/18 12:30:00	2025/07/18 12:30:00	DELETE

CompanyId	データポイント	シーケンス	システムシーケンス	オペレーション
A	XFv2	2025/07/18 12:30:00	2025/07/18 12:30:00	DELETE

出力：

CompanyId	データポイント	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	2025年7月18日 10時01分00秒	NULL	2025/07/18 10:05:00	2025年7月18日 12:20:00
A	XFv1	2025年7月18日 10時01分00秒	2025年7月18日 12時15分43秒	2025年7月18日 12:20:00	2025年7月18日 12:25:00
A	XFv1	2025年7月18日 10時01分00秒	2025/07/18 12:05:00	2025年7月18日 12:25:00	NULL
A	XFv3	2025/07/18 12:05:00	2025年7月18日 12時15分43秒	2025年7月18日 12:25:00	NULL
A	XFv2	2025年7月18日 12時15分43秒	NULL	2025年7月18日 12:20:00	2025/07/18 12:30:00
A	XFv2	2025年7月18日 12時15分43秒	2025/07/18 12:30:00	2025/07/18 12:30:00	NULL

CompanyId	データポイント	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	2025年7月18日 10時01分00秒	NULL	2025/07/18 10:05:00	2025年7月18日 12:20:00
A	XFv1	2025年7月18日 10時01分00秒	2025年7月18日 12時15分43秒	2025年7月18日 12:20:00	2025年7月18日 12:25:00
A	XFv1	2025年7月18日 10時01分00秒	2025/07/18 12:05:00	2025年7月18日 12:25:00	NULL
A	XFv3	2025/07/18 12:05:00	2025年7月18日 12時15分43秒	2025年7月18日 12:25:00	NULL
A	XFv2	2025年7月18日 12時15分43秒	NULL	2025年7月18日 12:20:00	2025/07/18 12:30:00
A	XFv2	2025年7月18日 12時15分43秒	2025/07/18 12:30:00	2025/07/18 12:30:00	NULL

XFv2 は 12:15:43 から終了時刻不明で有効であり、システムは 12:20:00～12:30:00 の間、そのように認識していました。削除が取り込まれた後、XFv2 は 12:30:00 までのみ有効であることが知られており、システム時間 12:30:00 から有効な修正履歴となります。

制限事項

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

その他のリソース

• 変更データ キャプチャとスナップショット: CDC概念、スナップショット、 SCDタイプについて学びます。
• AUTO CDCを使用して外部 RDBMS テーブルを複製します。 onceフローを使用して初期ハイドレーションを実行し、変更の処理を続行する方法を学習します。
• 高度な AUTO CDCトピック: AUTO CDCターゲットでの変更操作、変更データフィードの読み取り、メトリクスの処理について学びます。
• チュートリアル: チェンジデータキャプチャを使用してETLパイプラインを構築する