変更データキャプチャ (CDC) は、Delta テーブルに対する行レベルの挿入、更新、削除を増分イベントストリームとして記録します。ダウンストリームシステムは、このストリームを増分計算、データ同期、ストリーム処理、監査に利用できます。毎回テーブル全体をスキャンする必要はありません。
CDC は現在招待制プレビュー中です。使用方法については、「増分計算の概要」をご参照ください。
ユースケース
増分計算：変更レコードを使用してマテリアライズドビューを更新します。テーブル全体を再処理する必要はありません。
ストリーム処理：CDC レコードを Flink ジョブまたは他のストリームプロセッサーに送信します。
マルチエンジンデータ同期：異なるコンピューティングエンジン間で増分変更を伝播します。
ログ監査：コンプライアンスとガバナンスのために、すべての行レベル操作の完全な記録を保持します。
CDC モードの選択
CDC は 2 つのモードをサポートしています。ワークロードが Tunnel によるリアルタイム書き込みを必要とするかどうかに基づいて選択してください。
同期 CDC | 非同期 CDC | |
SQL DML 操作 | サポート | サポート |
Tunnel によるリアルタイム書き込み | 非サポート | サポート |
CDC データの可用性 | SQL 操作の完了直後 | 設定された間隔後に非同期で |
Delta テーブルでの CDC の有効化
同期 CDC
Delta テーブルを作成する際に、"acid.cdc.mode.enable" = "true" および "acid.cdc.build.async" = "false" プロパティを追加します。
CREATE TABLE acid_with_cdc_tbl (pk BIGINT NOT NULL PRIMARY KEY, val BIGINT)
tblproperties ("transactional" = "true",
"acid.cdc.mode.enable" = "true",
"acid.cdc.build.async" = "false");acid.cdc.mode.enable: After the Delta Table CDC feature is enabled, it runs asynchronously by default. CDC is built asynchronously when data changes are triggered by SQLDML operations, and CDC generation is also supported asynchronously in Tunnel real-time write scenarios.
acid.cdc.build.async：CDC データの構築方法を制御します。"false" に設定すると、同期 CDC テーブルが作成され、SQL DML 操作の完了直後に CDC データが生成されます。Tunnel によるリアルタイム書き込みはサポートされません。cdc.data.retain.hours：オプションのプロパティです。CDC データの保持期間 (時間単位) です。範囲：1～168。デフォルト値：24。CDC を有効にしたテーブルでは、INSERT INTO 操作はサポートされません。代わりに MERGE INTO を使用することを推奨します。
オプションのプロパティ:
プロパティ | 説明 | デフォルト |
| CDC データの保持期間 (時間単位) です。範囲：1～168。 |
|
非同期 CDC
Delta テーブルの作成時に、"acid.cdc.mode.enable"="true"、"acid.cdc.build.async"="true"、および "acid.cdc.build.interval"="300" を追加します。必要に応じて "cdc.data.retain.hours"="24" プロパティを追加できます。
CREATE TABLE acid_with_cdc_build_tbl (pk BIGINT NOT NULL PRIMARY KEY, val BIGINT)
tblproperties ("transactional" = "true",
"acid.cdc.mode.enable" = "true",
"acid.cdc.build.async" = "true",
"acid.cdc.build.interval" = "300");acid.cdc.mode.enable：Delta テーブルの CDC 機能を有効にすると、デフォルトで非同期で実行されます。SQL DML 操作によってデータ変更がトリガーされると、CDC は非同期で構築されます。また、Tunnel によるリアルタイム書き込みのシナリオにおいても、非同期での CDC 生成がサポートされます。Tunnel リアルタイム書き込みシナリオでは、DML 操作と CDC の生成も非同期にサポートされています。
acid.cdc.build.async：CDC の非同期構築を有効にします。Tunnel によるテーブルへのリアルタイム書き込みがサポートされ、SQL DML 操作に対しても CDC データが非同期で生成されます。acid.cdc.build.interval：非同期構築の間隔 (秒単位) です。有効値：[60-3540]。ビジネス要件や増分シナリオに応じてこのパラメーターを設定してください。cdc.data.retain.hours：オプションのプロパティです。CDC データの保持期間 (時間単位) です。範囲：1～168。デフォルト値：24。その他のオプションパラメーター (プロジェクトレベルまたはセッションレベル)：
odps.storage.orc.enable.memcmp.sort.key="true"、プロジェクトレベルで有効にすることを推奨します。これにより、CDC の非同期構築とクエリの両方のパフォーマンスが向上します。
CDC データのクエリ
table_changes 関数を使用して、バージョンまたはタイムスタンプ範囲で CDC 変更レコードをクエリします。
構文
SELECT * FROM table_changes('<table_name>', <start> [, <end>]);パラメーター
パラメーター | 必須 | タイプ | 説明 |
| はい | — | クエリ対象の Delta テーブル。 |
| はい | BIGINT または STRING | CDC データクエリの開始バージョン。バージョン番号は |
| いいえ | BIGINT または STRING | CDC データクエリの終了バージョン。省略した場合は、最新バージョンがデフォルトになります。STRING の場合は、 |
返される列
テーブルのデータ列に加えて、table_changes は 3 つのシステム列を返します：
列 | 説明 |
| 行が書き込まれたときのシステム時刻。 |
| 操作タイプ： |
| 行が更新の一部かどうか： |
__meta_op_type と __meta_is_update を組み合わせて、完全な操作を識別します：
|
| 操作 |
|
| 挿入による新しい行 |
|
| 更新後の値 (更新後の新しい値) |
|
| 更新前の値 (更新前の元の値) |
|
| 削除された行 |
例
ステップ 1: CDC を有効にしたテーブルを作成します。
CREATE TABLE acid_cdc_table(id1 STRING NOT NULL, id2 STRING NOT NULL, key1 BIGINT, key2 BIGINT, PRIMARY KEY(id1, id2))
tblproperties("transactional" = "true", "acid.cdc.mode.enable" = "true");ステップ 2: データを挿入します。
-- データ挿入時刻 2025-04-07 11:56:57
INSERT INTO acid_cdc_table VALUES ('1', '1006', 1006, 1006);
-- データ挿入時刻 2025-04-07 12:15:00
INSERT INTO acid_cdc_table VALUES ('1', '1008', 1008, 1008);
-- データ挿入時刻 2025-04-07 13:24:00
INSERT INTO acid_cdc_table VALUES ('1', '1032', 1032, 1032);
-- データ挿入時刻 2025-04-07 14:00:00
INSERT INTO acid_cdc_table VALUES ('1', '1045', 1045, 1045);
-- データ挿入時刻 2025-04-07 14:47:00
INSERT INTO acid_cdc_table VALUES ('1', '1045', 1045, 1045);ステップ 3: テーブルのバージョンを確認します。
SHOW HISTORY FOR TABLE acid_cdc_table;出力:
ObjectType ObjectId ObjectName VERSION(LSN) Time Operation
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000001 2025-04-07 11:55:59 CREATE
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000002 2025-04-07 11:56:57 APPEND
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000003 2025-04-07 12:00:13 MINOR_COMPACT
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000004 2025-04-07 12:15:32 APPEND
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000005 2025-04-07 12:30:02 MINOR_COMPACT
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000006 2025-04-07 13:24:47 APPEND
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000007 2025-04-07 13:30:02 MINOR_COMPACT
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000008 2025-04-07 14:00:41 APPEND
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000009 2025-04-07 14:15:15 MINOR_COMPACT
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000010 2025-04-07 14:47:46 APPEND
TABLE a4a78d3f6af04d85a57a90deff884021 acid_cdc_table 0000000000000011 2025-04-07 15:00:11 MINOR_COMPACTステップ 4: CDC レコードをクエリします。
2025-04-07 12:00:00 以降のすべての変更をクエリします (バージョン 3 と同等)：
SELECT * FROM table_changes('acid_cdc_table', '2025-04-07 12:00:00');
-- 以下と同等
SELECT * FROM table_changes('acid_cdc_table', 3);出力:
+------------+------------+------------+------------+------------------+----------------+------------------+
| id1 | id2 | key1 | key2 | __meta_timestamp | __meta_op_type | __meta_is_update |
+------------+------------+------------+------------+------------------+----------------+------------------+
| 1 | 1045 | 1045 | 1045 | 2025-04-07 14:00:34 | 1 | 0 |
| 1 | 1008 | 1008 | 1008 | 2025-04-07 12:15:28 | 1 | 0 |
| 1 | 1032 | 1032 | 1032 | 2025-04-07 13:24:43 | 1 | 0 |
| 1 | 1045 | 1045 | 1045 | 2025-04-07 14:47:41 | 1 | 0 |
+------------+------------+------------+------------+------------------+----------------+------------------+特定の範囲内の変更をクエリします (2025-04-07 12:00:00 から 13:30:00 まで、バージョン 3～6 と同等)：
SELECT * FROM table_changes('acid_cdc_table', '2025-04-07 12:00:00', '2025-04-07 13:30:00');
-- 以下と同等
SELECT * FROM table_changes('acid_cdc_table', 3, 6);出力:
+------------+------------+------------+------------+------------------+----------------+------------------+
| id1 | id2 | key1 | key2 | __meta_timestamp | __meta_op_type | __meta_is_update |
+------------+------------+------------+------------+------------------+----------------+------------------+
| 1 | 1008 | 1008 | 1008 | 2025-04-07 12:15:28 | 1 | 0 |
| 1 | 1032 | 1032 | 1032 | 2025-04-07 13:24:43 | 1 | 0 |
+------------+------------+------------+------------+------------------+----------------+------------------+ストリームオブジェクトを使用した CDC データの利用
ストリームオブジェクトは、Delta テーブルの CDC 履歴の読み取りオフセットを追跡します。"read_mode" = "cdc" を指定してストリームオブジェクトを作成し、CDC データを増分的に利用します。
ストリームオブジェクトの詳細については、「ストリームオブジェクト」をご参照ください。
構文
CREATE STREAM [IF NOT EXISTS] <stream_name>
ON TABLE <delta_table_name> VERSION AS OF <v>
strmproperties ("read_mode" = "cdc")"read_mode" = "cdc" プロパティは、ストリームオブジェクトがクエリバージョン範囲に基づいて CDC データを利用するように設定します。
例
-- CDC を有効にしたソーステーブルを作成します。
CREATE TABLE acid_with_cdc_stream (id1 BIGINT NOT NULL PRIMARY KEY, id2 BIGINT)
tblproperties ("transactional" = "true", "acid.cdc.mode.enable" = "true");
-- データを挿入します。
INSERT INTO acid_with_cdc_stream VALUES (1, 1006), (2, 1008), (3, 1032);
-- バージョン 1 から開始するストリームオブジェクトを作成します。
CREATE STREAM delta_table_stream ON TABLE acid_with_cdc_stream VERSION AS OF 1 strmproperties ("read_mode" = "cdc");
-- ストリームオブジェクトを確認します。
DESC STREAM delta_table_stream;出力:
Name delta_table_stream
Project yunqi_y****
Schema default
Create Time 2024-12-03 11:13:12
Last Modified Time 2024-12-03 11:13:12
Offset Version 1
Reference Table Project yunqi_y****
Reference Table Schema default
Reference Table Name acid_with_cdc_stream
Reference Table Id b89ec113f50944d5b8e52ce6a00c****
Reference Table Version 2
Parameters {"read_mode": "cdc"}