0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@kamo-shika
in

Databricks の AUTO CDC ターゲットから Change Data Feed を読んで、変わった行だけ AI 関数にかける

0
Last updated at Posted at 2026-10-17

各機能の挙動や制約は 2026 年 10 月時点のものです。利用前に最新のドキュメントを確認してください。

はじめに

ai_analyze_sentiment() や ai_query() などの AI 関数は、1 行ずつ基盤モデルを呼ぶので、行数に比例して費用がかかります。テーブル全体に毎回かけ直すのはもったいないので、本当に中身が変わった行だけにかけたくなります。

CDC で取り込んでいるデータだと、これが意外と難しいです。同じレコードが再送されてきたり、AI に関係ない列だけが更新されたりします。取り込んだ行をそのまま AI に流すと、変わっていない行にも AI が走ります。

そこで、AUTO CDC（旧 APPLY CHANGES）で作った silver テーブルの Change Data Feed（CDF） を読む構成を試しました。

以前は APPLY CHANGES のターゲットから CDF を読もうとすると、The operation is not supported on APPLY CHANGES Streaming Tables. というエラーで弾かれていました。2024 年 8 月の DLT リリース 2024.33 で読めるようになり、今は Databricks Runtime 15.2 以降なら、ほかの Delta テーブルと同じように読めます。地味ですが、めっちゃ便利なアップデートです。

この記事では、この構成を Lakeflow Spark Declarative Pipelines（旧 Delta Live Tables、以下 SDP）で組み、どの操作で AI が走るのかを実測しました。

Change Data Feed とは

Change Data Feed は、Delta テーブルの行ごとの変更の記録です。テーブルに delta.enableChangeDataFeed = true を付けておくと、それ以降の INSERT・UPDATE・DELETE が「どの行が・どう変わったか」として記録されます。

記録は、元の列に次の 3 つの列を足した形で読めます。

列 中身
_change_type 変更の種類。insert / update_preimage（更新前の値）/ update_postimage（更新後の値）/ delete
_commit_version 変更が入ったテーブルのバージョン
_commit_timestamp そのバージョンがコミットされた時刻

例として、今回の検証では silver に対して「id 2 の text を更新する」「id 6 を新しく追加する」という 2 つの操作をしました。そのとき silver の CDF には、次のように記録されていました。

silver のバージョン 3 からバージョン 4 で id 2 を更新し id 6 を追加すると、Change Data Feed には id 2 の更新前の値（update_preimage）と更新後の値（update_postimage）、id 6 の追加（insert）の 3 行だけが記録される

1 回の更新につき、更新前と更新後の 2 行が出ます。テーブル全体を読み直さなくても、ここを読めば「前回から何が変わったか」だけが分かる、というわけです。

読み方は 2 通りあります。SQL なら table_changes()、ストリームなら readChangeFeed オプションです。ストリームで読むと、前回どこまで読んだかを覚えているので、次の回は続きの変更だけが届きます。

SELECT * FROM table_changes('workspace.spk_cdf.spk_cdf_silver', 1);

spark.readStream.option("readChangeFeed", "true").table("workspace.spk_cdf.spk_cdf_silver")

DBR 19 以降には、delta.enableChangeDataFeed を付けなくても、行追跡（row tracking）が有効なテーブルなら読み取り時に変更を計算してくれる「自動 CDF」があります（2026 年 9 月に一般提供）。

試したところ、AUTO CDC のターゲットでも、従来の CDF を delta.enableChangeDataFeed: "false" で切って行追跡だけにした状態で、別のパイプラインから更新・削除・追加を読めました。ただし次の 2 点に注意が要ります。

  • 2026 年 10 月時点では、パイプラインの CURRENT チャネルは DBR 18.3 なので、読む側も含めて PREVIEW チャネル（DBR 19）にする必要がある。
  • DBR 19 では、AUTO CDC のターゲットに従来の CDF が自動でオンになる。自動 CDF にしたいときは、delta.enableChangeDataFeed: "false" を明示して切る。

自動 CDF そのものは、こちらの記事が詳しいです。

https://qiita.com/taka_yayoi/items/0dcb3a13e1598f237dd0

AUTO CDC と組み合わせると何がうれしいのか

AUTO CDC は、CDC イベントを受け取ってテーブルを最新の状態に保つ SDP の機能です。キー（今回は id）ごとに、版の番号（今回は seq 列）が一番新しいイベントだけを反映し、古い版や同じ版の再送は捨てます。

つまり AUTO CDC で作ったテーブル（ターゲット）の CDF には、捨てられた再送は出てこず、実際にテーブルが変わった行だけが出てきます。 これを AI 関数に渡せば、AI は本当に変わった行にしか走らないはずです。

構成

やりたいことは、silver の CDF から変わった行だけを一時ビューに出し、そこで 1 回だけ AI にかけることです。組んだ構成はこうなります。

bronze_src から取り込みパイプラインで bronze と silver（AUTO CDC、CDF 有効）を作り、分析パイプラインで silver の CDF を読んで、変わった行だけ AI 関数にかけた gold_enriched と、それを集計する gold_summary を作る構成

  • 取り込みパイプライン: 普通の Delta テーブル bronze_src を bronze に取り込み、AUTO CDC で silver を作ります。silver は CDF を有効にしておきます。
  • 分析パイプライン: silver の CDF から変わった行だけを読み、AI 関数で感情分析した gold_enriched を作ります。さらに、それを集計する MV gold_summary を作ります。

取り込み（bronze / silver）と分析（gold）でパイプラインを分けるのは、Databricks が推奨している形です。分析側が失敗しても取り込みは止まりませんし、更新の頻度が違うものを別々に回せます。分析パイプラインは CDF のどこまで読んだかを覚えているので、取り込みを何回か流してから分析を 1 回流しても、その間の変更をまとめて処理できます。

取り込みパイプライン

silver を AUTO CDC で作り、CDF を有効にします。

dp.create_streaming_table(
    name="spk_cdf_silver",
    table_properties={"delta.enableChangeDataFeed": "true"},
)

dp.create_auto_cdc_flow(
    target="spk_cdf_silver",
    source="spk_cdf_bronze",
    keys=["id"],
    sequence_by=col("seq"),
    stored_as_scd_type="1",
    apply_as_deletes=expr("op = 'D'"),
)
取り込みパイプラインの全文 
from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr


@dp.table(name="spk_cdf_bronze")
def spk_cdf_bronze():
    return spark.readStream.table("workspace.spk_cdf.spk_cdf_bronze_src")


dp.create_streaming_table(
    name="spk_cdf_silver",
    table_properties={
        "delta.enableChangeDataFeed": "true",
        "delta.enableRowTracking": "true",
    },
)

dp.create_auto_cdc_flow(
    target="spk_cdf_silver",
    source="spk_cdf_bronze",
    keys=["id"],
    sequence_by=col("seq"),
    stored_as_scd_type="1",
    apply_as_deletes=expr("op = 'D'"),
)

分析パイプライン

一時ビューで silver の CDF を読み、更新前の行（update_preimage）を落としてから AI 関数をかけます。AI を呼ぶのはこの一時ビューの 1 か所だけです。

(
    spark.readStream.option("readChangeFeed", "true")
    .table("workspace.spk_cdf.spk_cdf_silver")
    .where("_change_type IN ('insert', 'update_postimage', 'delete')")
    .withColumn("sentiment", expr("ai_analyze_sentiment(text)"))
    .withColumn("enriched_at", expr("current_timestamp()"))
)

これを gold_enriched に AUTO CDC で書きます。順序には silver のバージョン _commit_version を使い、delete が来たら gold からも消します。enriched_at は AI を呼んだ時刻で、この値が変わった行が AI が走った行です。

集計の MV は refresh_policy="incremental_strict" にしました。増分で更新できないときは、黙って全件を再計算せずにエラーで止まります。AI 関数は MV ではなく一時ビュー側で呼んでいます。AI 関数は結果が毎回同じとは限らない（非決定的な）関数なので、MV の中で呼ぶと増分更新の対象にならず、毎回全行で推論が走るおそれがあるためです。

分析パイプラインの全文 
from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

SILVER = "workspace.spk_cdf.spk_cdf_silver"


@dp.temporary_view(name="spk_cdf_b_silver_changes")
def spk_cdf_b_silver_changes():
    return (
        spark.readStream.option("readChangeFeed", "true")
        .table(SILVER)
        .where("_change_type IN ('insert', 'update_postimage', 'delete')")
        .withColumn("sentiment", expr("ai_analyze_sentiment(text)"))
        .withColumn("enriched_at", expr("current_timestamp()"))
    )


dp.create_streaming_table(
    name="spk_cdf_b_gold_enriched",
    table_properties={"delta.enableRowTracking": "true"},
)

dp.create_auto_cdc_flow(
    target="spk_cdf_b_gold_enriched",
    source="spk_cdf_b_silver_changes",
    keys=["id"],
    sequence_by=col("_commit_version"),
    stored_as_scd_type="1",
    apply_as_deletes=expr("_change_type = 'delete'"),
    except_column_list=["_change_type", "_commit_timestamp"],
)


@dp.materialized_view(name="spk_cdf_b_gold_summary", refresh_policy="incremental_strict")
def spk_cdf_b_gold_summary():
    return spark.sql(
        "SELECT sentiment, status, count(*) AS n "
        "FROM spk_cdf_b_gold_enriched GROUP BY sentiment, status"
    )

試した結果

bronze_src（id, text, status, seq, op の 5 列。seq が版の番号、op = 'D' が削除）に行を足して CDC イベントの代わりにし、取り込み → 分析の順に update を流して、そのたびに enriched_at と MV の更新方法を記録しました。MV の更新方法は、イベントログの planning_information で確かめられます。

MV の更新方法を確かめる SQL 
SELECT timestamp, message
FROM event_log('<分析パイプラインの ID>')
WHERE event_type = 'planning_information'
ORDER BY timestamp;
Run 操作 AI が走った行 MV の更新方法
1 id 1〜4, 6 が silver にある状態で、分析パイプラインを初めて流す 全行 COMPLETE_RECOMPUTE（初回）
2 id 7 を新規、id 2 の text を変更、id 3 を同じ seq・同じ内容で再送 2, 7 GROUP_AGGREGATE（増分）
3 id 4 の status だけを変更（text は同じ） 4 GROUP_AGGREGATE（増分）
4 id 1 を同じ内容のまま、seq だけ上げて再送 1 GROUP_AGGREGATE（増分）
5 id 6 を削除 （6 が gold から消えた） GROUP_AGGREGATE（増分）

silver で変わった行だけが gold に流れ、AI もその行にしか走りませんでした。 Run 2 で同じ版を再送した id 3 は、silver の AUTO CDC が捨てるので CDF にも出てきません。削除も gold まで届き、MV は 2 回目以降ずっと増分で更新されています。

Run 2 の gold_enriched はこうなりました。id 2 と 7 だけ enriched_at が新しくなっています。

id  text                                       status  sentiment  enriched_at
1   I love this product                        open    positive   2026-10-17T03:24:06.854Z
2   Now it keeps crashing, I regret buying it  open    negative   2026-10-17T03:26:32.909Z
3   It is okay I guess                         open    neutral    2026-10-17T03:24:06.854Z
4   Absolutely fantastic service               closed  positive   2026-10-17T03:24:06.854Z
6   Pretty good value for money                open    positive   2026-10-17T03:24:06.854Z
7   Shipping was slow but support was kind     open    mixed      2026-10-17T03:26:32.909Z

なお、分析パイプラインの初回は CDF を開始バージョンなしで読んだので、silver のその時点の中身が全部 insert として届きます。初回だけは全行に AI が走ります。

一方で、気になる点も 2 つ残りました。

  • Run 3 では status しか変えていないのに、id 4 に AI が走った。CDF は行単位なので、どの列が変わったかまでは見ていない。
  • Run 5 の削除でも、消す行に ai_analyze_sentiment() を呼んでいる。

AI の入力が変わったときだけ AI を呼ぶ

この 2 点を直した版も作りました。考え方は「同じ text の AI の結果は使い回す」です。

  • text のハッシュ（SHA-256）を作り、AI の結果をハッシュごとに別のテーブルに貯める。
  • 一定期間内に同じハッシュが来たら AI を呼ばず、貯めた結果を使う。
  • 削除行は AI にかけない。

「一定期間内に同じハッシュが来たか」を覚えておくのに、ストリーム処理のウォーターマークを使います。

ウォーターマークとは

ストリームで重複を見分けるには、「このハッシュはもう見た」という記録をどこかに持っておく必要があります。Spark のストリーム処理では、これを状態（state）としてチェックポイントに保存します。

何もしないと、この記録は消えずに増え続けます。text の種類が 100 万あれば 100 万件を覚えておくことになり、処理のたびに読み書きされる状態がどんどん重くなります。

ウォーターマークは、「データの時刻で見て、これより古いものはもう来ない（来ても扱わない）」という線です。withWatermark("時刻の列", "ウォーターマークの期間") と書くと、届いたデータの最新の時刻からウォーターマークの期間ぶん前に線が引かれ、それより古い記録は状態から消えます。

たとえばウォーターマークの期間が 30 日で、届いたデータの最新の時刻が 10 月 31 日なら、線は 10 月 1 日に引かれます。10 月 1 日より前の記録は消え、それ以降の記録だけが残ります。

データの時刻の軸の上で、届いたデータの最新の時刻からウォーターマークの期間ぶん前にウォーターマークの線が引かれる。線より古い text A の記録は状態から消え、線より新しい text B の記録は残る。text A がまた来たら AI をもう一度呼び、text B がまた来たら AI を呼ばない

重複排除と組み合わせる dropDuplicatesWithinWatermark を使うと、「ウォーターマークの期間内に同じキーが来たら落とす。期間を過ぎて記録が消えた後に来たら、新しいものとして通す」という動きになります。

今回の使い方

時刻の列には、CDF の _commit_timestamp（silver にコミットされた時刻）を使いました。ウォーターマークの期間は、本番なら 30 日などを想定しています。

(
    changes  # silver の変更行（削除行は除く）に text_hash を足したもの
    .withWatermark("_commit_timestamp", AI_CACHE_TTL)
    .dropDuplicatesWithinWatermark(["text_hash"])
    .withColumn("sentiment", expr("ai_analyze_sentiment(text)"))
)

こうすると、AI の呼び出しは次のように決まります。

  • ウォーターマークの期間内に同じ text が来た（status だけの変更や再送） → 落ちるので AI を呼ばない。
  • ウォーターマークの期間を過ぎてから同じ text が来た → 記録が消えているので、AI をもう一度呼ぶ。
  • 新しい text が来た → AI を呼ぶ。

ウォーターマークの期間を過ぎると同じ text でも AI がもう一度走るので、AI の結果のテーブルは AUTO CDC でハッシュをキーにして上書きし、ハッシュごとに 1 行を保ちます。変更行には、このテーブルからハッシュで結果を引いてくっつけます。

silver の変更行に text のハッシュを付け、ai_results では期間内はハッシュごとに 1 回だけ AI を呼ぶ（削除行は除外）。変更行にはハッシュで AI の結果を引いてくっつけ、gold_enriched と gold_summary に流す構成

ウォーターマーク版の分析パイプラインの全文 
from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

SILVER = "workspace.spk_cdf.spk_cdf_silver"
# 同じ text の AI の結果を使い回す期間
AI_CACHE_TTL = spark.conf.get("spk.ai_cache_ttl", "30 days")


@dp.temporary_view(name="spk_cdf_d_silver_changes")
def spk_cdf_d_silver_changes():
    return (
        spark.readStream.option("readChangeFeed", "true")
        .table(SILVER)
        .where("_change_type IN ('insert', 'update_postimage', 'delete')")
        .withColumn("text_hash", expr("sha2(text, 256)"))
    )


# 期間内に同じ text が来たら落とし、期間を過ぎたらもう一度 AI にかける。削除行は AI にかけない
@dp.temporary_view(name="spk_cdf_d_ai_runs")
def spk_cdf_d_ai_runs():
    return (
        spark.readStream.table("spk_cdf_d_silver_changes")
        .where("_change_type != 'delete'")
        .withWatermark("_commit_timestamp", AI_CACHE_TTL)
        .dropDuplicatesWithinWatermark(["text_hash"])
        .select("text_hash", "text", "_commit_version")
        .withColumn("sentiment", expr("ai_analyze_sentiment(text)"))
        .withColumn("analyzed_at", expr("current_timestamp()"))
        .drop("text")
    )


# AI の結果はハッシュごとに最新 1 行にする（期限切れで分析し直したら上書き）
dp.create_streaming_table(
    name="spk_cdf_d_ai_results",
    table_properties={"delta.enableRowTracking": "true"},
)

dp.create_auto_cdc_flow(
    target="spk_cdf_d_ai_results",
    source="spk_cdf_d_ai_runs",
    keys=["text_hash"],
    sequence_by=col("_commit_version"),
    stored_as_scd_type="1",
    except_column_list=["_commit_version"],
)


# 変更行に AI の結果をくっつける（ストリームと静的テーブルの join）。ここでは AI を呼ばない
@dp.temporary_view(name="spk_cdf_d_enriched_changes")
def spk_cdf_d_enriched_changes():
    ai = spark.read.table("spk_cdf_d_ai_results")
    return spark.readStream.table("spk_cdf_d_silver_changes").join(ai, "text_hash", "left")


dp.create_streaming_table(
    name="spk_cdf_d_gold_enriched",
    table_properties={"delta.enableRowTracking": "true"},
)

dp.create_auto_cdc_flow(
    target="spk_cdf_d_gold_enriched",
    source="spk_cdf_d_enriched_changes",
    keys=["id"],
    sequence_by=col("_commit_version"),
    stored_as_scd_type="1",
    apply_as_deletes=expr("_change_type = 'delete'"),
    except_column_list=["_change_type", "_commit_timestamp", "text_hash"],
)


@dp.materialized_view(name="spk_cdf_d_gold_summary", refresh_policy="incremental_strict")
def spk_cdf_d_gold_summary():
    return spark.sql(
        "SELECT sentiment, status, count(*) AS n FROM spk_cdf_d_gold_enriched GROUP BY sentiment, status"
    )

結果

ウォーターマークの期間を 10 分に縮めて試しました（パイプラインの設定で spk.ai_cache_ttl を渡しています）。AI の呼び出し回数は、AI の結果のテーブルに書かれた行数で数えています。

操作 AI の呼び出し 理由
初回（5 行） 5 回 すべて初めての text
id 4 の status だけ変更、id 9 を新規 1 回（id 9 だけ） id 4 の text はウォーターマークの期間内に分析済み
2 分後: id 9 と id 4 の status だけ変更 1 回（id 4 だけ） id 9 はウォーターマークの期間内。id 4 は初回から 10 分以上たち、記録が消えていた
id 7 を削除、id 2 の text を変更 1 回（id 2 だけ） 削除行は AI にかけない

status だけの変更と削除では AI が呼ばれず、ウォーターマークの期間を過ぎた text だけがもう一度分析されました。 AI の結果のテーブルは、分析し直した後もハッシュごとに 1 行のままです。MV も、2 回目以降はずっと増分で更新されています。

ここで 1 つ注意があります。ウォーターマークの期限は実際の経過時間ではなく、届いたデータの時刻で進みます。新しい変更が届いたときに線が引き直され、古い記録が消えます。変更が何も来なければ、時間がたっても記録は消えません。

ウォーターマークの期間の長さは、使い方しだいです。

  • 長くするほど AI の呼び出しは減るが、ストリームの状態は大きくなる。
  • 短くするほど、同じ text でも定期的に分析し直される。モデルを新しくしたときに古い結果を入れ替えたいなら、むしろ都合がいい。
  • すぐに全部やり直したいときは、分析パイプラインを full refresh する。

使うときに気をつけること

silver を full refresh すると、AI の分析も全部やり直しになる

分析パイプラインは silver の CDF を続きから読んでいます。取り込みパイプラインで silver を作り直すと、分析パイプラインは続きを読めなくなり、こちらも full refresh することになります。そうなると、全行に AI が走り直します。

誤って作り直さないように、silver には pipelines.reset.allowed=false を付けておくと安心です。

CDF は後から付けられる

最初に CDF を付け忘れても大丈夫でした。CDF なしで作った bronze に、後から table_properties={"delta.enableChangeDataFeed": "true"} を足して普通の update を流したら、full refresh なしで有効になりました。ストリームも続きから動いています。

ただし、付ける前の変更の記録は当てにしないほうが安全です。今回の bronze は追記だけのテーブルだったので、付ける前の分も insert として読めました。それでも version 0 から読もうとすると、DELTA_MISSING_CHANGE_DATA で止まりました。更新や削除がある silver のようなテーブルでは、付けた後の分から読む前提にしておくのがよさそうです。

まとめ

AUTO CDC のターゲットの CDF を読んで、変わった行だけ AI 関数にかける構成を試しました。押さえておきたいのは次の 3 つです。

  • AUTO CDC が捨てた重複や古い版は CDF に出ない。 silver の CDF を読めば、AI は silver で変わった行にしか走らない。集計の MV も増分で更新できる。
  • 取り込みと分析はパイプラインを分ける。 分析パイプラインは CDF の続きから読むので、更新の頻度が違っても困らない。
  • AI の入力列が変わったときだけ呼ぶなら、AI の結果をハッシュごとに貯める。 重複の判定はウォーターマークで期間を区切る。status だけの変更や削除では AI が呼ばれず、ウォーターマークの期間を過ぎた text だけが分析し直される。

AI 関数の費用は呼んだ行数に比例するので、呼ぶ行をこうして絞れれば、そのぶん費用を抑えられそうです。

参考リンク

https://docs.databricks.com/aws/ja/tables/features/change-data-feed

https://docs.databricks.com/aws/ja/ldp/cdc

https://docs.databricks.com/aws/ja/ldp/cdc-advanced

https://docs.databricks.com/aws/ja/release-notes/dlt/2024/33/

https://docs.databricks.com/aws/ja/ldp/best-practices

https://docs.databricks.com/aws/ja/structured-streaming/watermarks

https://docs.databricks.com/aws/ja/ldp/incremental-refresh

https://docs.databricks.com/aws/ja/ldp/properties

https://docs.databricks.com/aws/ja/large-language-models/ai-functions

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?