Azure Databricksで変更データフィードを使用する

変更データフィード (CDF) は、Delta Lake テーブルまたは Apache Iceberg v3 テーブルのバージョン間の行レベルの変更を追跡します。

Azure Databricks では、次の 2 つの方法がサポートされています。

自動変更データフィード: 行系列メタデータを使用して、テーブルの読み取り中に変更を計算します。これには個々のテーブル構成は必要ありません。Delta Lake および Apache Iceberg v3 テーブルで動作します。自動変更データフィードを参照してください。
レガシー変更データフィード: テーブルへの書き込み時に変更を記録します。 Delta Lake テーブルのみをサポートします。個々のテーブル構成が必要です。 Delta Lake のレガシ変更データフィードを参照してください。

変更データフィードは、次のような一般的なデータユースケースに使用できます。

最後のパイプライン実行以降に変更された行のみを処理する増分 ETL パイプライン。
コンプライアンスとガバナンスの要件に関するデータ変更を追跡する監査証跡。
ダウンストリームテーブル、キャッシュ、または外部システムへの変更を同期するデータレプリケーションワークロード。

自動変更データフィード

重要

この機能はパブリックプレビュー段階です。ワークスペース管理者は、[ プレビュー] ページからこの機能へのアクセスを制御できます。 Manage Azure Databricks プレビューを参照してください。

自動変更データフィードは、Delta Lake の行追跡と Apache Iceberg v3 の行系列を使用して、書き込み時ではなく、クエリ時に行レベルの変更を計算します。従来の変更データフィードとは異なり、自動変更データフィードは個々のテーブル構成を必要とせず、Delta Lake テーブルと Apache Iceberg v3 テーブルで動作します。

変更は、 MERGE INTO 操作と UPDATE 操作の書き込みごとに計算されるわけではないため、自動変更データフィードを使用すると、従来の変更データフィードと比較して書き込みパフォーマンスが向上し、ストレージコストが削減されます。

自動変更データフィードは、従来の変更データフィードと同じ table_changes() および readChangeFeed API を使用し、バッチクエリ、構造化ストリーミング、Databricks から Databricks Delta Lake Sharing と連携します。バッチクエリ内の変更の読み取りおよび変更データの増分処理を参照してください。

要件

Databricks Runtime 18 以降
Unity カタログに登録されているサポートされているテーブル形式:
- 行追跡が有効になっている Delta Lake 形式または Iceberg v3 形式のマネージドテーブル。
- 行追跡が有効になっている Delta Lake 形式の外部テーブル。

Databricks Unity カタログのテーブルの種類に関する説明を参照してください。

Note

変更データフィードは Apache Iceberg 仕様の一部ではありません。Azure Databricksリーダーは Apache Iceberg v3 テーブルの自動変更データフィードに対してクエリを実行できますが、外部 Iceberg リーダーではクエリを実行できません。 Iceberg テーブルの仕様を参照してください。

Delta Lake の場合、自動変更データフィードに対してクエリを実行できるのは、Azure Databricks リーダーだけです。

変更データフィードを使用する

変更データフィードを使用するには、要件を満たすテーブルを使用していることを確認します。要件を参照してください。

変更データフィードをバッチ読み取りするには、次の操作を行います。

Python

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("<table_name>")

Scala

spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .table("<table_name>")

SQL

SELECT * FROM table_changes('<table_name>', 0)

変更データフィードのバッチ読み取りの詳細については、バッチクエリでの変更の読み取りを参照してください。

読み取り変更データフィードをストリーミングするには、次の操作を行います。

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .table("<table_name>")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .table("<table_name>")

変更データフィードのストリーミング読み取りの詳細については、「変更データの増分処理」を参照してください。

従来の変更データフィードから移行する

Delta Lake テーブルを従来の変更データフィードから自動変更データフィードに移行するには、次の操作を行います。

テーブルが要件を満たしていることを確認します。
次のコマンドを実行して、従来の変更データフィードをオフにします。

ALTER TABLE <table_name> UNSET TBLPROPERTIES ('delta.enableChangeDataFeed');

レガシと自動の両方の変更データフィードを一緒に使用することはできません。

データフィードスキーマを変更する

テーブルの変更データフィードから読み取ると、クエリでは最新バージョンのテーブルのスキーマが使用されます。 Azure Databricksではスキーマの変更と進化のほとんどの操作がサポートされますが、列マッピングを持つテーブルには制限があります。列マッピングを含むテーブルを参照してください。

Delta Lake テーブルのスキーマのデータ列に加えて、変更データフィードには、変更イベントの種類を識別するメタデータ列が含まれています。

列名	タイプ	Values
`_change_type`	String	次のものが含まれます: `insert`、 `update_preimage`、 `update_postimage`、 `delete`。 `preimage` は更新前の値で、 `postimage` は更新後の値です。
`_commit_version`	Long	含まれるもの: その変更を含む Delta ログまたはテーブルバージョン。
`_commit_timestamp`	Timestamp	内容: コミットの作成時に関連付けられたタイムスタンプ。

スキーマにこれらのメタデータ列と同じ名前の列が含まれている場合、テーブルで変更データフィードを使用することはできません。変更データフィードを有効にする前に、テーブル内の列の名前を変更して、この競合を解決します。

変更データを段階的に処理する

Databricks では、構造化ストリーミングと組み合わせて変更データフィードを使用して、テーブルからの変更を増分処理することをお勧めします。テーブルの変更データフィードのバージョンを自動的に追跡するには、Azure Databricks の Structured Streaming を使用する必要があります。 SCD タイプ 1 またはタイプ 2 のテーブルを使用した CDC 処理については、「 AUTO CDC API: パイプラインを使用して変更データキャプチャを簡略化する」を参照してください。

ストリームが最初に開始されると、変更データフィードはテーブルの最新のスナップショットを INSERT レコードとして返し、その後の変更を変更データとして返します。変更データフィードは、変更データと新しいデータ行の両方を同時にテーブルトランザクションログにコミットします。

テーブルの変更データフィードを読み取るようにストリームを構成するには、オプション readChangeFeed を次のように true に設定します。

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .table("myTable")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .table("myTable")

レート制限

Azure Databricksでは、変更データを読み取る際のレート制限 (maxFilesPerTrigger、maxBytesPerTrigger) とexcludeRegexがサポートされます。ストリーミング Delta Lake オプションの完全な一覧については、「 Delta Lake」を参照してください。

必要に応じて、開始バージョンを指定できます。「開始バージョンの指定」を参照してください。開始スナップショット以外のバージョンの場合、レート制限はコミット全体にアトミックに適用されます。現在のバッチにコミット全体が含まれているか、現在のバッチがコミットを次のバッチに延期します。

リプレイテーブルの履歴

変更データフィードは、テーブルに対するすべての変更の永続的なレコードとして機能することを意図したものではありません。変更データフィードが有効になった後に発生した変更のみを記録します。新しいストリーミング読み取りを開始して、現在のバージョンとその後のすべての変更をキャプチャできます。

変更データフィード内のレコードは一時的なものであり、指定された保持期間でのみアクセスできます。トランザクションログでは、テーブルのバージョンとそれに対応する変更データフィードのバージョンが一定の間隔で削除されます。バージョンが削除されると、そのバージョンの変更データフィードを読み取ることができなくなりました。

永続的な履歴の変更データをアーカイブする

ユースケースでテーブルに対するすべての変更の永続的な履歴を保持する必要がある場合は、増分ロジックを使用して変更データフィードから新しいテーブルにレコードを書き込みます。

次の例では、 trigger.AvailableNow を使用して、監査または完全な変更の再生のために使用可能なデータをバッチワークロードとして処理する方法を示します。

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .table("source_table")
  .writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(availableNow=True)
  .toTable("target_table")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .table("source_table")
  .writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(Trigger.AvailableNow)
  .toTable("target_table")

開始バージョンを指定する

特定の時点からの変更を読み取るために、タイムスタンプまたはバージョン番号を使用して開始バージョンを指定します。バッチ読み取りには、開始バージョンが必要です。必要に応じて、終了バージョンを指定して範囲を制限できます。テーブル履歴の詳細については、「タイムトラベル」を参照してください。

変更データフィードを使用する構造化ストリーミングワークロードを構成する場合、開始バージョンを指定すると処理のパフォーマンスに影響する可能性があります。

新しいデータ処理パイプラインは、通常、ストリームの初回起動時にテーブル内のすべての既存のレコードを INSERT 操作として記録する既定の動作の恩恵を受けます。
ターゲットテーブルに、特定の時点までの適切な変更を含むすべてのレコードが既に含まれている場合は、ソーステーブルの状態を INSERT イベントとして処理しないように、開始バージョンを指定します。

次の例は、チェックポイントが破損しているストリーミングエラーから復旧する方法を示しています。この例では、次のような条件を想定します。

テーブル作成時にソーステーブルで変更データフィードが有効になりました。
ターゲットダウンストリームテーブルは、バージョン 75 までの変更をすべて処理しました。
ソーステーブルのバージョン履歴は、バージョン 70 以降で使用できます。

既存のターゲットテーブルへの書き込みストリームを定義するときは、新しいチェックポイントの場所を指定する必要があります。

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")
  .writeStream
  .option("checkpointLocation", "<new-checkpoint-path>")
  .toTable("target_table")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")
  .writeStream
  .option("checkpointLocation", "<new-checkpoint-path>")
  .toTable("target_table")

重要

開始バージョンを指定し、そのバージョンがテーブル履歴で使用できない場合、ストリームは新しいチェックポイントから開始できません。マネージドテーブルは履歴バージョンを自動的にクリーンアップするため、指定したすべての開始バージョンが最終的に削除されます。

テーブル履歴の再生を参照してください。

バッチクエリの変更を読み取る

バッチクエリ構文を使用すると、特定のバージョンから始まるすべての変更を読み取ったり、指定したバージョン範囲内の変更を次のように読み取ったりできます。

バージョンを整数として指定し、タイムスタンプを文字列として yyyy-MM-dd[ HH:mm:ss[.SSS]]形式で指定します。
開始バージョンと終了バージョンは含まれています。開始バージョンから最新バージョンに読み取る場合は、開始バージョンのみを指定します。
変更データフィードが有効になる前にバージョンを指定すると、エラーが発生します。

開始バージョンオプションと終了バージョンオプションを使用してバッチ読み取りを行うには、次の操作を行ってください。

SQL

バージョン 0 から 10に読み取る場合は、次の操作を行います。

SELECT * FROM table_changes('tableName', 0, 10)

2 つのタイムスタンプバージョン間で読み取りを行うには、次の操作を行います。

--
SELECT * FROM table_changes('tableName', '2021-04-21 05:45:46', '2021-05-21 12:00:00')

開始バージョンから最新バージョンへの読み取りを行うには、次の操作を行います。

SELECT * FROM table_changes('tableName', 0)

名前に特殊文字が含まれるテーブルの変更を読み取るために、次の操作を行います。

SELECT * FROM table_changes('`schema`.`dotted.tableName`', '2021-04-21 06:45:46', '2021-05-21 12:00:00')

テーブル値関数table_changes参照してください。

Python

バージョン 0 から 10に読み取る場合は、次の操作を行います。

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .option("endingVersion", 10) \
  .table("myDeltaTable")

2 つのタイムスタンプ間で読み取りを行うには、次の操作を行います。

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingTimestamp", '2021-04-21 05:45:46') \
  .option("endingTimestamp", '2021-05-21 12:00:00') \
  .table("myDeltaTable")

開始バージョンから最新バージョンへの読み取りを行うには、次の操作を行います。

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("myDeltaTable")

Scala

バージョン 0 から 10に読み取る場合は、次の操作を行います。

spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .option("endingVersion", 10)
  .table("myDeltaTable")

2 つのタイムスタンプ間で読み取りを行うには、次の操作を行います。

spark.read
  .option("readChangeFeed", "true")
  .option("startingTimestamp", "2021-04-21 05:45:46")
  .option("endingTimestamp", "2021-05-21 12:00:00")
  .table("myDeltaTable")

開始バージョンから最新バージョンへの読み取りを行うには、次の操作を行います。

spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .table("myDeltaTable")

範囲外のバージョンを処理する

既定では、最後のコミットを超えるバージョンまたはタイムスタンプを指定すると、クエリはエラー timestampGreaterThanLatestCommitを返します。

Databricks Runtime 11.3 LTS 以降では、次のように範囲外のバージョンの許容範囲を有効にすることができます。

SET spark.databricks.delta.changeDataFeed.timestampOutOfRange.enabled = true;

この構成を有効にすると、クエリは次のように異なる結果を返します。

最後のコミットを超える開始バージョンまたはタイムスタンプは、空の結果を返します。
最後のコミットを超える終了バージョンまたはタイムスタンプは、最初から最後のコミットまでのすべての変更を返します。

Delta Lake の従来の変更データフィード

従来の変更データフィードでは、個々の Delta Lake テーブルを手動で構成する必要があります。変更データフィードは Apache Iceberg 仕様に含まれていないため、Apache Iceberg テーブルはサポートされていません。 Databricks では、自動変更データフィードに移行することをお勧めします。「従来の変更データフィードからの移行」を参照してください。

レガシ変更データフィードが有効になっている場合、ランタイムレコードは、テーブルに書き込まれるすべてのデータの 変更イベント を記録します。これには、指定した行が挿入、削除、または更新されたかどうかを示すメタデータと共に行データが含まれます。

従来の変更データフィードでは、自動変更データフィードと同じ readChangeFeed と table_changes() 読み取り API が使用されます。バッチクエリでの変更データの増分処理と変更の読み取りを参照してください。

従来の変更データフィードを有効にする

個々のテーブルでレガシ変更データフィードを明示的に有効にする必要があります。以下のいずれかの方法を使用します。

新しいテーブル

delta.enableChangeDataFeed = true コマンドで table プロパティCREATE TABLEを設定します。

CREATE TABLE student (id INT, name STRING, age INT)
  TBLPROPERTIES (delta.enableChangeDataFeed = true)

Note

任意の期間、レガシ変更データフィードを無効にしてから再度有効にした場合、その間隔はクエリできません。間隔中に変更を照会するには、自動変更データフィードを使用します。自動変更データフィードを参照してください。

既存のテーブル

delta.enableChangeDataFeed = true コマンドで table プロパティALTER TABLEを設定します。

ALTER TABLE myDeltaTable
  SET TBLPROPERTIES (delta.enableChangeDataFeed = true)

ストレージに関する考慮事項

マネージドテーブルはデータの変更を効率的に記録し、他の機能を使用してストレージレイアウトを最適化する場合があります。

従来の変更データフィードでは、次のストレージ動作を考慮する必要があります。

変更が別々のファイルに記録される可能性があるため、ストレージコストが少し増加する可能性があります。
挿入のみの削除やパーティション全体の削除など、一部の操作では変更データファイルが生成されません。 Azure Databricksは、トランザクションログから直接変更データフィードを計算します。
変更データファイルでは、テーブルの保持ポリシーが使用されます。 VACUUM コマンドは変更データファイルを削除し、トランザクションログからの変更はチェックポイント保持ポリシーを使用します。

Databricks では、変更データファイルに直接クエリを実行して変更データフィードを再構築しないことをお勧めします。常に Delta Lake API と Apache Iceberg API を使用します。

制限事項

変更データフィードには、次の制限事項を考慮してください。

列マッピングを含むテーブル

Delta Lake テーブルで列マッピングを有効にすると、データファイルを書き換えずに列を削除または名前変更できます。「Delta Lake 列マッピングを使用して列の名前を変更および削除する」を参照してください。

ただし、変更データフィードには、非追加スキーマの変更後の制限があります。非追加スキーマの変更には、次の操作が含まれます。

列の名前を変更または削除します。
列のデータ型を変更します。
ALTER COLUMN ... SET NOT NULLなど、列の null 許容を変更します。制約NOT NULL参照してください。

非加法スキーマ変更が発生したトランザクションまたは範囲の変更データフィードを読み取ることはできません。

指定したバッチ読み取りの範囲の前または後に非加法スキーマの変更を許可するために、クエリでは、最新のテーブルバージョンではなく、範囲の終了バージョンのスキーマが使用されます。バージョン範囲が非加法スキーマ変更にまたがる場合でも、クエリは失敗します。

自動変更データフィード

変更データフィードは Apache Iceberg 仕様ではサポートされていないため、外部の Iceberg クライアントは自動変更データフィードに対してクエリを実行できません。 Iceberg テーブルの仕様を参照してください。
複数ステートメントトランザクションの場合、トランザクション中にソーステーブルが変更された場合、自動変更データフィードはサポートされません。
行フィルターまたは列マスクを含むテーブルでは、自動変更データフィードはサポートされていません。「行フィルターと列マスク」を参照してください。
変更データフィードクエリは、列の名前変更、削除、データ型の変更など、追加以外のスキーマ変更が発生したテーブルバージョンにまたがることはできません。スキーマ変更の前後の範囲にクエリを分割します。

フィードバック

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

Last updated on 2026-06-23

Azure Databricksで変更データ フィードを使用する

自動変更データフィード

要件

変更データフィードを使用する

Python

Scala

SQL

Python

Scala

従来の変更データ フィードから移行する

データ フィード スキーマを変更する

変更データを段階的に処理する

Python

Scala

レート制限

リプレイテーブルの履歴

永続的な履歴の変更データをアーカイブする

Python

Scala

開始バージョンを指定する

Python

Scala

バッチ クエリの変更を読み取る

SQL

Python

Scala

範囲外のバージョンを処理する

Delta Lake の従来の変更データ フィード

従来の変更データ フィードを有効にする

新しいテーブル

既存のテーブル

ストレージに関する考慮事項

制限事項

列マッピングを含むテーブル

自動変更データフィード

フィードバック

その他のリソース

Azure Databricksで変更データフィードを使用する

従来の変更データフィードから移行する

データフィードスキーマを変更する

バッチクエリの変更を読み取る

Delta Lake の従来の変更データフィード

従来の変更データフィードを有効にする