AUTO CDC の高度なトピック

基本的な AUTO CDC および AUTO CDC FROM SNAPSHOT API 以外にも、ターゲット テーブルで DML を実行したり、CDC ターゲットから変更データ フィードを読み取ったり、処理メトリックを監視したり、部分的な更新を適用したり、一口ストレージを使用して変更を追跡したりできます。 AUTO CDC API の概要については、「AUTO CDC API: パイプラインを使用して変更データ キャプチャを簡略化する」を参照してください。

ターゲット ストリーミング テーブル内のデータを追加、変更、または削除する

パイプラインでテーブルを Unity Catalog に発行する場合は、挿入、更新、削除、マージステートメントなどの データ操作言語 (DML) ステートメントを使用して、 AUTO CDC ... INTO ステートメントによって作成されたターゲット ストリーミング テーブルを変更できます。

パイプラインは、指定した SEQUENCE BY 列を使用し、ターゲット テーブルの __START_AT 列と __END_AT 列に適切なシーケンス値を伝達するため (SCD 型 2 の場合)、DML ステートメントでこれらの列の有効な値を使用して、レコードの適切な順序を維持する必要があります。 AUTO CDC のしくみを参照してください。

ストリーミング テーブルで DML ステートメントを使用する方法の詳細については、「ストリーミング テーブルのデータの追加、変更、または削除」を参照してください。

次の例では、開始シーケンスが 5 のアクティブ なレコードを挿入します。

INSERT INTO my_streaming_table (id, name, __START_AT, __END_AT) VALUES (123, 'John Doe', 5, NULL);

CREATE VIEW my_employees_view AS
SELECT
  *,
  __START_AT AS valid_from,
  __END_AT AS valid_to
FROM my_scd2_target_table;

AUTO CDC ターゲット テーブルから変更データ フィードを読み取る

Databricks Runtime 15.2 以降では、他の Delta テーブルから変更データ フィードを読み取るのと同じ方法で、 AUTO CDC クエリまたは AUTO CDC FROM SNAPSHOT クエリのターゲットであるストリーミング テーブルから変更データ フィードを読み取ることができます。 ターゲット ストリーミング テーブルから変更データ フィードを読み取る場合は、次のものが必要です。

• ターゲット ストリーミング テーブルは、Unity カタログに発行する必要があります。 「パイプラインで Unity カタログを使用する」を参照してください。
• ターゲット ストリーミング テーブルから変更データ フィードを読み取る場合は、Databricks Runtime 15.2 以降を使用する必要があります。 別のパイプラインで変更データ フィードを読み取るために、Databricks Runtime 15.2 以降を使用するようにパイプラインを構成する必要があります。

他の Delta テーブルから変更データ フィードを読み取るのと同じ方法で、Lakeflow パイプラインで作成されたターゲット ストリーミング テーブルから変更データ フィードを読み取ります。 Python および SQL の例を含む Delta Change Data Feed 機能の使用方法の詳細については、Azure Databricks で Change Data Feed を使用するを参照してください。

マテリアライズドビューから変更データフィードを読み取る

LakeflowパイプラインやDatabricks SQLで作成されたマテライズドビューから変更データフィードを読み取ることができます。 これを使って、Azure Databricks外の宛先へのマテリアル化されたビュー変更をレプリケートしたり、監査や報告のためにマテリアル化されたビューの変更履歴を保持したりします。

マテリアル化されたビューは自動変更データフィードを使用するため、変更データフィード自体をオンにすることはありません。 代わりに、必要な各具体化されたビューで変更データフィードを有効にすると、以下の要件を満たす必要があります。 自動変更データ フィードを参照してください。

• 変更データフィードを読むには、Databricks Runtime 18 LTS以上、クラシックコンピュート、サーバーレスコンピュート、またはDatabricks SQLを使用する必要があります。

• マテリアライズされたビュー、それを作成するパイプライン、またはそれを読み込むパイプラインは PREVIEW チャネルを使用しなければなりません。

• マテリアライズドビューは行追跡が有効でなければなりません。 サーバーレスコンピュート上のマテリアルズビューはデフォルトで行追跡が有効です。 Azure Databricksの行追跡を参照してください。 マテリアライズド・ビューで行トラッキングが有効になっているかどうかを確認するには、次を実行します:

  SHOW TBLPROPERTIES my_mv ('delta.enableRowTracking');
  

• マテライズドビューからの変更データフィードを読み取るには、パイプラインまたはマテリアライズドビューで外部メタデータフラグを有効にしてください。 手順については「 データセットへのアクセスを有効にする方法」をご覧ください。

マテリアライズドビューからの変更データフィードは、他のDeltaテーブルと同じ方法で、 table_changes() 関数、ストリーミングリード、または readChangeFeed オプションを使って読み取ります。 SQLやPythonの構文や例については、「Azure DatabricksでChange data feedを使う」をご覧ください。

Databricks SQLのマテリアル化されたビューやストリーミングテーブルの内部から、マテリアル化されたビュー変更データフィードを読み取ることができます:

CREATE OR REFRESH STREAMING TABLE sales
  AS SELECT * FROM STREAM my_mv WITH (readChangeFeed=true)

制限事項

自動変更データフィードの制限に加え、マテリアライズドビューから変更データフィードを読む場合、以下の制限が適用されます。

• 変更データフィードには、マテリアル化されたビューが完全に書き換えられた際に変更されていない行が含まれ、同じ行の複数の更新を単一のイベントに統合することはありません。 これらを除外するには、変更データフィードをすべての列でグループ化して集約し、同じ行の値を持つ挿入と削除を特定します。
• マテリアライズされたビューのために変更データフィードをクエリできるのはAzure Databricksだけです。 外部のデルタレイクおよびアイスバーグのクライアントはできません。
• Lakeflowパイプライン内では、別のパイプラインからのマテリアル化されたビュー変更データフィードのみを読み取ることができ、そのパイプラインは PREVIEW チャネルを使用しなければなりません。 マテリアライズドビューの作成と同じパイプライン内での変更データフィードを読み取ることはサポートされていません。
• 物質化されたビューからベクトル検索インデックスを作成することはできません。

パイプライン内の CDC クエリによって処理されたレコードに関するデータを取得する

次のメトリックは、 AUTO CDC クエリによってキャプチャされます。

• num_upserted_rows: 更新中にデータセットにアップサートされた出力行の数。
• num_deleted_rows: 更新中にデータセットから削除された既存の出力行の数。

num_output_rowsメトリックは非CDCフローの出力であり、AUTO CDCクエリではキャプチャされません。

部分的な更新プログラムを適用する

ソースが変更された列のみを送信する場合、 AUTO CDC は、変更レコードに含まれず、ターゲット値を変更せずに残す必要がある列と、 nullに明示的に設定されている列を区別する必要があります。これは、ターゲット値を null で上書きする必要があります。 既定では、 IGNORE NULL UPDATES はすべての null を "更新しない" マーカーとして扱うので、明示的な nullを適用できません。 このあいまいさを解決するには、次の 3 つの方法のいずれかを選択します。

COLUMNS TO UPDATE を IGNORE NULL UPDATES と組み合わせることはできず、バイテンポラル テーブルではサポートされていません。

経験則として、プロデューサーが各レコードで変更された列を認識し、その情報をソース列に格納できる場合 (複数のプロデューサーが同じソースに書き込む場合や、更新可能な列のセットが時間の経過と同時に増加する場合など) は、 COLUMNS TO UPDATE を選択します。 パイプライン所有者が更新可能な列の固定セットを事前に認識し、パイプライン コードでそれらを制御する場合は、 IGNORE NULL UPDATES ON を選択します。

次の例では、 columnsToUpdate という名前のソース列を使用して、レコードの更新を変更する各列 (明示的に null に設定された列を含む) を制御します。

Python

from pyspark import pipelines as dp

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "sequenceNum",
  stored_as_scd_type = 1,
  columns_to_update = "columnsToUpdate"
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  target
FROM
  stream(cdc_source)
KEYS
  (id)
SEQUENCE BY
  sequenceNum
STORED AS
  SCD TYPE 1
COLUMNS TO UPDATE
  columnsToUpdate;

完全なパラメーター リファレンスについては、「 AUTO CDC INTO (パイプライン) 」と 「create_auto_cdc_flow」を参照してください。

Bitemporal AUTO CDC

SCD タイプ1とタイプ2は単一時制であり、単一の時間軸に沿った変更を追跡します。 Bitemporal は SCD Type 2 の履歴を拡張して、2 つの時間ディメンション間の変更を追跡し、2 つのパースペクティブを区別します。

• 営業時間: イベントが実際に発生したとき。
• システム時刻: システムがイベントを記録または取り込んだ時刻。

SCD タイプ 2 と同様に、bitemporal はレコードの完全な履歴を保持します。 2 つ目のタイムラインが追加されるため、データの表示内容と、過去の任意の時点でシステムが信じていた内容の両方を再構築できます。

たとえば、ヘッジファンドは、ソースシステムから株式データを取り込みます。 アクメ社の株価は1月1日に変更されるが、ファンドはその更新を1月5日まで取り込まない。 Bitemporal AUTO CDCを使用すると、Acme Corpの実際の株価は1月1日(営業時間)と、ファンドが1月3日(システム時間)に取引決定を行ったときにシステムが信じた価格という2つの異なる質問に答えます。 これらのタイムラインを区別する機能は、監査、規制レポート、財務上の意思決定に役立ちます。

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

Bitemporal AUTO CDC の例

次の例では、少数の合成 CDC イベントを使って、二重時間ターゲット テーブルを作成します。 bt列には業務時間があり、st列にはシステム時刻があります。

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

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

次の一連の変更は、二時制テーブルが1社に対する挿入、更新、順序外の更新、および削除をどのように記録するかを示しています。 シーケンス列は __START_AT 列と __END_AT 列 (業務時間) 列を生成し、システムシーケンス列は __SYSTEM_START_AT 列と __SYSTEM_END_AT (システム時刻) 列を生成します。

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

変更 1: 挿入

A 社は 2025 年 7 月 18 日 10:01:00 (営業時間) に追加されますが、10:05:00 (システム時刻) まで取り込まれません。

入力：

Output:

XFv1 は 10:01:00 以降で有効であり、既知の終了はありません。 システムはこの事実をシステム時刻 10:05:00 に学習し、終了は不明です。

変更 2: 更新

A 社は 2025 年 7 月 18 日 12:15:43 (営業時間) に更新され、システムは 12:20:00 (システム時間) にイベントを使用します。 システムは、更新プログラムが認識される前に信じていたものと、更新プログラムが取り込まれた後の修正されたビジネス履歴の両方を保持します。

入力：

Output:

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 年 7 月 18 日 12:05:00（業務時刻）に更新されたことを示す順序外の更新が到着しますが、12:25:00（システム時刻）まで取り込まれません。 更新がシステム時間上では後から到着しても、ビジネス時間ではそれより前の時点を示す場合、システムは過去のビジネス時間の履歴を訂正し、順不同の更新が到着する前に保持していた内容と、訂正後の履歴の両方を保持します。

入力：

Output:

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 つの行に表示され、会社が存在しなくなったときとシステムが削除を知ったときの両方の完全な監査証跡が保持されます。

入力：

Output:

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

パイプラインでの CDC 処理に使用されるデータ オブジェクトは何ですか?

Hive メタストアでターゲット テーブルを宣言すると、次の 2 つのデータ構造が作成されます。

• ターゲット テーブルに割り当てられた名前を使用するビュー。
• CDC 処理を管理するためにパイプラインによって使用される内部バッキング テーブル。 このテーブルの名前は、ターゲット テーブル名に対するプリペンド __apply_changes_storage_ によって指定されます。

たとえば、 dp_cdc_targetという名前のターゲット テーブルを宣言すると、 dp_cdc_target という名前のビューとメタストアに __apply_changes_storage_dp_cdc_target という名前のテーブルが表示されます。 ビューにクエリを実行して、処理されたデータにアクセスします。 バッキング テーブルを直接変更しないでください。