高度なAUTO CDCトピック
基本的な AUTO CDCと AUTO CDC FROM SNAPSHOT APIsを超えて、ターゲットテーブルでDMLをランし、CDCターゲットからチェンジデータフィードを読み取り、処理メトリクスを監視し、部分的な更新を適用できます。バイテンポラルストレージについては、「 バイテンポラルAUTO CDC」を参照してください。AUTO CDC APIsの概要については、「 「AUTO CDC APIs : パイプラインによるチェンジデータキャプチャの簡素化」」を参照してください。
ターゲットのストリーミングテーブルのデータを追加、変更、または削除します。
パイプラインがテーブルをUnity Catalogに公開する場合、insert、update、delete、merge ステートメントを含むデータ操作言語(DML) ステートメントを使用して、 AUTO CDC ... INTOステートメントによって作成されたターゲット ストリーミング テーブルを変更できます。
- ストリーミングテーブルのテーブルスキーマを変更する DML ステートメントはサポートされていません。DML ステートメントによってテーブルスキーマの進化が発生しないようにしてください。
- ストリーミングテーブルを更新する DML ステートメントは、Databricks Runtime 13.3 LTS 以上を使用する共有Unity Catalog クラスターまたはSQL ウェアハウスでのみ実行できます。
- ストリーミングには追加専用のデータ ソースが必要なため、処理で (DML ステートメントなどによる) 変更を伴うソース ストリーミング テーブルからのストリーミングが必要な場合は、ソース ストリーミング テーブルの読み取り時にSkipChangeCommits フラグを設定します。
skipChangeCommitsが設定されている場合、ソース テーブルのレコードを削除または変更するトランザクションは無視されます。処理にストリーミング テーブルが必要ない場合は、ターゲット テーブルとしてマテリアライズドビュー (追加のみの制限がない) を使用できます。
パイプラインは指定された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);
SCD タイプ 2 ターゲット テーブル内の__START_AT列と__END_AT列の名前を変更する必要がある場合 (たとえば、ダウンストリーム スキーマの要件に一致させるため)、ターゲット テーブルにビューを作成します。
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クエリのターゲットであるDeltaテーブルから変更データフィードを読み取ることができます。 ターゲットのストリーミング テーブルから変更データフィードを読み取るには、次のものが必要です。
- ターゲット ストリーミングテーブルは Unity Catalogにパブリッシュする必要があります。 「パイプラインで Unity Catalog を使用する」を参照してください。
- ターゲット ストリーミング テーブルから変更データフィードを読み取るには、 Databricks Runtime 15.2 以降を使用する必要があります。 別のパイプラインで変更データフィードを読み取るには、 Databricks Runtime 15.2 以降を使用するようにパイプラインを構成する必要があります。
LakeFlow Pipelinesで作成されたターゲットストリーミングテーブルからチェンジデータフィードを読み取る方法は、他の Delta テーブルからチェンジデータフィードを読み取るのと同じ方法です。Delta チェンジデータフィード機能の使用方法 (Python や SQL の例を含め) の詳細については、Databricks でチェンジデータフィードを使用する」を参照してください。
変更データフィード レコードには、変更イベントのタイプを識別するメタデータが含まれています。 テーブル内のレコードが更新されると、関連付けられた変更レコードのメタデータには通常、 update_preimageおよびupdate_postimageイベントに設定された_change_type値が含まれます。
ただし、主キー値の変更を含む更新がターゲット ストリーミング テーブルに行われた場合、 _change_type値は異なります。 変更に主キーの更新が含まれる場合、 _change_typeメタデータ フィールドはinsertおよびdeleteイベントに設定されます。主キーの変更は、 UPDATEまたはMERGEステートメントを使用してキー フィールドの 1 つに手動で更新が行われたとき、または SCD タイプ 2 テーブルの場合は、 __start_atフィールドが以前の開始シーケンス値を反映するように変更されたときに発生する可能性があります。
AUTO CDCクエリは、SCD タイプ 1 と SCD タイプ 2 の処理で異なる主キー値を決定します。
SCD Type | Primary key |
|---|---|
SCD type 1, and the pipelines Python interface | The primary key is the value of the |
SCD type 2 | The primary key is the |
マテリアライズドビューからのチェンジデータフィードの読み取り
ベータ版
この機能はベータ版です。
LakeflowパイプラインまたはDatabricks SQLで作成されたマテリアライズドビューから、チェンジデータフィードを読み取ることができます。これを使用して、マテリアライズドビューの変更をDatabricks外部の宛先にレプリケートしたり、監査やレポートのためにマテリアライズドビューの変更履歴を保持したりします。
マテリアライズドビューは自動チェンジデータフィードを使用するため、チェンジデータフィード自体を有効にする必要はありません。その代わりに、以下の要件を満たすことで、必要な各マテリアライズドビューでチェンジデータフィードを有効にします。自動チェンジデータフィードを参照してください。
-
チェンジデータフィードを読み取るには、クラシックコンピュート、Serverlessコンピュート、またはDatabricks SQL上で、Databricks Runtime 18 LTS以降を使用する必要があります。
-
マテリアライズドビュー、それを作成するパイプライン、またはそれを読み取るパイプラインは、
PREVIEWチャンネルを使用する必要があります。 -
マテリアライズドビューでは、行追跡が有効になっている必要があります。Serverless Compute上のマテリアライズドビューでは、defaultで行追跡が有効になっています。Databricks の行追跡を参照してください。マテリアライズドビューで行追跡が有効になっているかどうかを確認するには、以下を実行します：
SQLSHOW TBLPROPERTIES my_mv ('delta.enableRowTracking'); -
マテリアライズドビューからチェンジデータフィードを読み取るには、パイプラインまたはマテリアライズドビューで外部メタデータフラグを有効にします。手順については、データセットへのアクセスを有効にする方法を参照してください。
マテリアライズドビューからのチェンジデータフィードの読み取りは、他のDeltaテーブルの場合と同様に、table_changes()関数、ストリーミング読み取り、またはreadChangeFeedオプションを使用して行います。SQL および Python の構文と例については、Databricksでのチェンジデータフィードの使用を参照してください。
Databricks SQL マテリアライズドビューまたはストリーミングテーブル内から、マテリアライズドビューのチェンジデータフィードを読み取ることができます：
CREATE OR REFRESH STREAMING TABLE sales
AS SELECT * FROM STREAM my_mv WITH (readChangeFeed=true)
制限事項
自動チェンジデータフィードの制限に加え、マテリアライズドビューからチェンジデータフィードを読み取る際には、以下が適用されます。
- マテリアライズドビューが完全に書き換えられる場合、チェンジデータフィードには変更されていない行が含まれます。また、同じ行に対する複数の更新を単一のイベントに統合することはありません。これらを除外するには、すべての列でグループ化してチェンジデータフィードを集計し、同じ行値を共有する挿入と削除を特定します。
- Databricks のみがマテリアライズドビューのチェンジデータフィードをクエリーできます。外部の Delta Lake クライアントおよび Iceberg クライアントは実行できません。
- LakeFlow Pipelines 内では、別のパイプラインからのみマテリアライズドビューのチェンジデータフィードを読み取ることができ、そのパイプラインは
PREVIEWチャンネルを使用する必要があります。マテリアライズドビューを作成するのと同じパイプライン内で、そのマテリアライズドビューのチェンジデータフィードを読み取ることはサポートされていません。 - マテリアライズドビューからベクトル検索インデックスを作成することはできません。
パイプライン内の CDC クエリによって処理されたレコードに関するデータを取得する
次のメトリクスは、 AUTO CDCクエリによってのみキャプチャされ、 AUTO CDC FROM SNAPSHOTクエリによってはキャプチャされません。
次のメトリクスは、 AUTO CDCクエリによってキャプチャされます。
num_upserted_rows: 更新中にデータセットにアップサートされた出力行の数。num_deleted_rows: 更新中にデータセットから削除された既存の出力行の数。
非 CDC フローの出力であるnum_output_rowsメトリクスは、 AUTO CDCクエリではキャプチャされません。
部分的な更新を適用します
ソースが変更された列のみを送信する場合、AUTO CDCは、変更レコードに存在しない列（ターゲット値を変更せずに残すべき）と、nullに明示的に設定された列（ターゲット値をnullで上書きすべき）を区別する必要があります。By default, IGNORE NULL UPDATESはすべてのnullを「更新しない」マーカーとして扱うため、明示的なnullを適用できません。この曖昧さを解決するには、次の3つの方法のいずれかを選択します。
手法 | 使用方法 | 挙動 |
|---|---|---|
| 小さく固定された列のセットは | 一覧表示されている列は、入力値が |
| ほとんどの列では | リストされた列は、明示的な |
| 各変更レコードは異なる列セットを更新するか、更新可能な列セットが時間の経過とともに変化します。 | ソース列は、変更レコードごとに更新する列を指定します。リストされた列は、明示的な |
COLUMNS TO UPDATE IGNORE NULL UPDATESと組み合わせることはできず、バイテンポラルテーブルではサポートされていません。
経験則として、プロデューサーが各レコードでどの列が変更されたかを認識しており、その情報をソース列で伝達できる場合（複数のプロデューサーが同じソースに書き込む場合や、更新可能な列のセットが時間とともに増加する場合など）は、COLUMNS TO UPDATEを選択してください。パイプラインの所有者が更新可能な列の固定セットを事前に認識しており、パイプラインコードでそれらを制御することを好む場合は、IGNORE NULL UPDATES ONを選択してください。
次の例では、columnsToUpdateという名前のソース列を使用して、nullに明示的に設定された列を含む、各変更レコードが更新する列を制御します：
- Python
- SQL
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"
)
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 」を参照してください。
パイプラインの CDC 処理に使用されるデータ オブジェクトは何ですか?
Hive metastore でターゲットテーブルを宣言すると、2 つのデータ構造が作成されます。
- ターゲットテーブルに割り当てられた名前を使用するビュー。
- CDC 処理を管理するためにパイプラインによって使用される内部バッキング テーブル。このテーブルの名前は、ターゲット テーブル名の前に
__apply_changes_storage_を付加して付けられます。
たとえば、 dp_cdc_targetという名前のターゲット テーブルを宣言すると、メタストアにdp_cdc_targetという名前のビューと__apply_changes_storage_dp_cdc_targetという名前のテーブルが表示されます。処理されたデータにアクセスするには、ビューをクエリします。バッキングテーブルを直接変更しないでください。
これらのデータ構造はAUTO CDC処理にのみ適用され、 AUTO CDC FROM SNAPSHOT処理には適用されません。これらは、 Unity Catalogではなく、 Hive metastoreにのみ適用されます。