Delta Change Data Feed完全ガイド｜CDC・増分処理・下流連携

Change Data Feed（CDF）は、Delta Tableに対する変更（INSERT / UPDATE / DELETE）を自動的に記録し、下流のパイプラインやシステムに「何が変わったか」を効率的に伝播させるための機能です。Silver → Gold 間の増分処理や、外部システムへのデータ連携に活用されます。

この記事では、CDFの有効化、_change_typeの種類と意味、バッチ/ストリーミングでの読み取り方法、foreachBatch + MERGEでの下流反映パターン、Lakeflow Spark Declarative Pipelines (旧 Delta Live Tables) APPLY CHANGES INTOとの関係、そして試験で問われるポイントを整理します。

CDFの有効化

CDFはテーブル単位で有効化します。新規テーブルの作成時、または既存テーブルのプロパティ変更で設定できます。CDF有効化後の変更のみが記録されます。有効化前の変更履歴は参照できません。

-- 新規テーブル作成時に有効化
CREATE TABLE silver.orders (
  order_id BIGINT,
  customer_id BIGINT,
  amount DECIMAL(10,2),
  status STRING,
  updated_at TIMESTAMP
)
TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- 既存テーブルに対して有効化
ALTER TABLE silver.orders
SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- ワークスペース全体でデフォルト有効化
SET spark.databricks.delta.properties.defaults.enableChangeDataFeed = true;

_change_type の種類

CDFで読み取ったデータには、自動的に3つのメタカラムが追加されます。 最も重要なのは _change_type で、変更の種類を示します。

UPDATEでは preimage（変更前）と postimage（変更後）の2行が記録されます。 「何から何に変わったか」を完全にトレースできるのがCDFの強みです。 追加のメタカラムとして _commit_version（Deltaバージョン番号）と_commit_timestamp（コミット時刻）も付与されます。

CDFの読み取り

バッチ読み取り: table_changes() 関数

-- バージョン範囲で変更を取得
SELECT * FROM table_changes('silver.orders', 5, 10);

-- タイムスタンプ範囲で変更を取得
SELECT * FROM table_changes('silver.orders', '2026-03-01', '2026-03-27');

-- 特定の_change_typeだけフィルタ
SELECT * FROM table_changes('silver.orders', 5)
WHERE _change_type IN ('insert', 'update_postimage');

ストリーミング読み取り: readChangeFeed オプション

# バージョン指定で開始
cdf_stream = (spark.readStream
  .format("delta")
  .option("readChangeFeed", "true")
  .option("startingVersion", 5)
  .table("silver.orders")
)

# タイムスタンプ指定で開始
cdf_stream = (spark.readStream
  .format("delta")
  .option("readChangeFeed", "true")
  .option("startingTimestamp", "2026-03-01")
  .table("silver.orders")
)

下流パイプラインへの増分反映（foreachBatch + MERGE）

CDFの最も一般的な使い方は、Silver → Gold 間の増分処理です。Gold層の集約テーブルを毎回フルリビルドする代わりに、CDFで変更分だけを取得してMERGEで反映します。

def update_gold(batch_df, batch_id):
    updates = batch_df.filter(
        "_change_type IN ('insert', 'update_postimage')"
    )
    deletes = batch_df.filter("_change_type = 'delete'")

    updates.createOrReplaceTempView("updates")
    deletes.createOrReplaceTempView("deletes")

    # UPSERT: 新規挿入と更新後の状態を反映
    spark.sql("""
      MERGE INTO gold.order_summary t
      USING updates s ON t.order_id = s.order_id
      WHEN MATCHED THEN UPDATE SET *
      WHEN NOT MATCHED THEN INSERT *
    """)

    # DELETE: 削除された行をGoldからも削除
    spark.sql("""
      MERGE INTO gold.order_summary t
      USING deletes s ON t.order_id = s.order_id
      WHEN MATCHED THEN DELETE
    """)

(spark.readStream
  .format("delta")
  .option("readChangeFeed", "true")
  .table("silver.orders")
  .writeStream
  .foreachBatch(update_gold)
  .option("checkpointLocation", "/checkpoint/gold_orders")
  .trigger(availableNow=True)
  .start()
)

trigger(availableNow=True)はバッチ的にストリームを実行し、 利用可能なすべてのCDFデータを処理した後にストリームを自動停止します。 スケジュールジョブとの相性が良いパターンです。

外部システムへの連携

CDFは外部システムへの増分データ連携にも使えます。全データの再送信ではなく、変更分だけを送るのでネットワークと処理コストを大幅に削減できます。

• Snowflake / Redshift への増分ロード
• Elasticsearch / OpenSearch へのインデックス更新
• Redis / DynamoDB へのキャッシュ更新
• Kafka への変更イベント発行

Lakeflow Spark Declarative Pipelines APPLY CHANGES INTOとの関係

Lakeflow Spark Declarative PipelinesのAPPLY CHANGES INTOは、CDFと類似の概念ですが レイヤーが異なります。CDFはDelta Table側の「変更記録機能」であり、 APPLY CHANGES INTOはDLTパイプライン側の「CDC取り込み構文」です。

-- Lakeflow Spark Declarative Pipelines APPLY CHANGES INTO の例
APPLY CHANGES INTO LIVE.silver_orders
FROM STREAM(LIVE.bronze_orders_cdc)
KEYS (order_id)
SEQUENCE BY updated_at
COLUMNS * EXCEPT (_rescued_data)
STORED AS SCD TYPE 1;

保持期間とVACUUM

CDFデータはDelta Tableの _change_data ディレクトリに保存されます。 VACUUMを実行すると、保持期間を超えたCDFデータも削除されます。

-- VACUUM実行（7日間保持）
VACUUM silver.orders RETAIN 168 HOURS;

-- 下流パイプラインの処理遅延を考慮して保持期間を長めに設定
ALTER TABLE silver.orders
SET TBLPROPERTIES (
  'delta.deletedFileRetentionDuration' = 'interval 30 days'
);

試験で問われるポイント

• CDFの有効化方法: TBLPROPERTIES (delta.enableChangeDataFeed = true)
• _change_type の4種類: insert / update_preimage / update_postimage / delete
• readChangeFeed オプションでのストリーミング読み取り
• table_changes() 関数でのバッチ読み取り
• CDFが有効化前の変更は取得できないこと
• UPDATE時にpreimageとpostimageの2行が記録されること
• APPLY CHANGES INTOはDLTパイプライン用の構文（CDFとは別レイヤー）

問題で確認

よくある質問