Change Data Capture (CDC) mencatat operasi INSERT, UPDATE, dan DELETE tingkat baris pada tabel Delta sebagai aliran event inkremental. Sistem downstream dapat mengonsumsi aliran ini untuk komputasi inkremental, sinkronisasi data, pemrosesan aliran, dan auditing—tanpa perlu memindai seluruh tabel pada setiap eksekusi.
CDC saat ini berada dalam pratinjau undangan. Untuk petunjuk penggunaan, lihat Ikhtisar perhitungan inkremental.
Kasus penggunaan
Komputasi inkremental: Konsumsi catatan perubahan untuk memperbarui Tampilan yang di-materialisasi tanpa memproses ulang seluruh tabel.
Pemrosesan aliran: Masukkan catatan CDC ke dalam Pekerjaan Flink atau prosesor aliran lainnya.
Sinkronisasi data multi-engine: Sebarkan perubahan inkremental di berbagai Mesin komputasi.
Auditing log: Simpan catatan lengkap semua operasi tingkat baris untuk kepatuhan dan tata kelola.
Pilih mode CDC
CDC mendukung dua mode. Pilih berdasarkan apakah workload Anda memerlukan penulisan real-time melalui Tunnel.
Synchronous CDC | Asynchronous CDC | |
Operasi DML SQL | Didukung | Didukung |
Penulisan Real-time melalui Tunnel | Tidak didukung | Didukung |
Ketersediaan data CDC | Segera setelah operasi SQL selesai | Secara asinkron, setelah interval yang dikonfigurasi |
Aktifkan CDC pada tabel Delta
Synchronous CDC
Tambahkan properti "acid.cdc.mode.enable"="true" dan "acid.cdc.build.async" = "false" saat membuat tabel Delta.
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: Setelah fitur CDC Delta Table diaktifkan, fitur ini berjalan secara asinkron secara default. CDC dibuat secara asinkron ketika perubahan data dipicu oleh SQL.Operasi DML, dan pembuatan CDC juga didukung secara asinkron dalam skenario penulisan real-time Tunnel.
acid.cdc.build.async: Mengontrol cara data CDC dibuat. Saat diatur ke "false", tabel CDC sinkron dibuat, dan data CDC dihasilkan segera setelah operasi DML SQL selesai. Penulisan real-time melalui Tunnel tidak didukung.cdc.data.retain.hours: Properti opsional. Periode retensi data CDC, dalam jam. Rentang: 1–168. Nilai default: 24.Tabel dengan CDC yang diaktifkan tidak mendukung operasi INSERT INTO. Kami menyarankan Anda menggunakan MERGE INTO sebagai gantinya.
Properti opsional:
Property | Description | Default |
| Periode retensi data CDC, dalam jam. Rentang: 1–168. |
|
Asynchronous CDC
Tambahkan "acid.cdc.mode.enable"="true", "acid.cdc.build.async"="true", dan "acid.cdc.build.interval"="300" saat membuat tabel Delta. Secara opsional, Anda dapat menambahkan properti "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: Setelah fitur CDC Delta Table diaktifkan, fitur ini berjalan secara asinkron secara default. CDC dibuat secara asinkron saat perubahan data dipicu oleh SQL.Operasi DML, dan pembuatan CDC juga didukung secara asinkron dalam skenario penulisan real-time Tunnel.
acid.cdc.build.async: Mengaktifkan pembuatan CDC secara asinkron. Penulisan real-time ke tabel melalui Tunnel didukung, dan data CDC juga dihasilkan secara asinkron untuk operasi DML SQL.acid.cdc.build.interval: Interval pembuatan asinkron, dalam detik. Nilai valid: [60–3540]. Konfigurasikan parameter ini berdasarkan skenario bisnis atau inkremental.cdc.data.retain.hours: Properti opsional. Periode retensi data CDC, dalam jam. Rentang: 1–168. Nilai default: 24.Parameter opsional lainnya (tingkat Proyek atau tingkat session):
odps.storage.orc.enable.memcmp.sort.key="true". Kami menyarankan mengaktifkannya di tingkat Proyek, yang meningkatkan kinerja pembuatan CDC asinkron maupun kueri.
Kueri data CDC
Gunakan fungsi table_changes untuk mengkueri catatan perubahan CDC berdasarkan versi atau rentang waktu.
Sintaksis
SELECT * FROM table_changes('<table_name>', <start> [, <end>]);Parameter
Parameter | Required | Type | Description |
| Ya | — | Tabel Delta yang akan dikueri. |
| Ya | BIGINT atau STRING | Versi awal untuk kueri data CDC. Temukan nomor versi dengan |
| Tidak | BIGINT atau STRING | Versi akhir untuk kueri data CDC. Jika tidak ditentukan, nilai default-nya adalah versi terbaru. Untuk STRING, gunakan format |
Kolom yang dikembalikan
Selain kolom data tabel, fungsi table_changes mengembalikan tiga kolom sistem:
Column | Description |
| Waktu sistem saat baris ditulis. |
| Jenis operasi: |
| Apakah baris tersebut merupakan bagian dari operasi UPDATE: |
Gabungkan __meta_op_type dan __meta_is_update untuk mengidentifikasi operasi lengkap:
|
| Operation |
|
| Baris baru dari operasi INSERT |
|
| Nilai setelah UPDATE (nilai baru setelah operasi UPDATE) |
|
| Nilai sebelum UPDATE (nilai asli sebelum operasi UPDATE) |
|
| Baris yang dihapus |
Contoh
Langkah 1: Buat tabel dengan CDC yang diaktifkan.
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");Langkah 2: Masukkan data.
-- Waktu penyisipan data 2025-04-07 11:56:57
INSERT INTO acid_cdc_table VALUES ('1', '1006', 1006, 1006);
-- Waktu penyisipan data 2025-04-07 12:15:00
INSERT INTO acid_cdc_table VALUES ('1', '1008', 1008, 1008);
-- Waktu penyisipan data 2025-04-07 13:24:00
INSERT INTO acid_cdc_table VALUES ('1', '1032', 1032, 1032);
-- Waktu penyisipan data 2025-04-07 14:00:00
INSERT INTO acid_cdc_table VALUES ('1', '1045', 1045, 1045);
-- Waktu penyisipan data 2025-04-07 14:47:00
INSERT INTO acid_cdc_table VALUES ('1', '1045', 1045, 1045);Langkah 3: Temukan versi tabel.
SHOW HISTORY FOR TABLE acid_cdc_table;Output:
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_COMPACTLangkah 4: Kueri catatan CDC.
Kueri semua perubahan setelah 2025-04-07 12:00:00 (setara dengan versi 3):
SELECT * FROM table_changes('acid_cdc_table', '2025-04-07 12:00:00');
-- Setara dengan
SELECT * FROM table_changes('acid_cdc_table', 3);Output:
+------------+------------+------------+------------+------------------+----------------+------------------+
| 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 |
| 2 | 1045 | 1045 | 1045 | 2025-04-07 14:47:41 | 1 | 0 |
+------------+------------+------------+------------+------------------+----------------+------------------+Kueri perubahan dalam rentang tertentu (2025-04-07 12:00:00 hingga 13:30:00, setara dengan versi 3–6):
SELECT * FROM table_changes('acid_cdc_table', '2025-04-07 12:00:00', '2025-04-07 13:30:00');
-- Setara dengan
SELECT * FROM table_changes('acid_cdc_table', 3, 6);Output:
+------------+------------+------------+------------+------------------+----------------+------------------+
| 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 |
+------------+------------+------------+------------+------------------+----------------+------------------+Konsumsi data CDC dengan Stream
Objek Stream melacak offset baca dalam riwayat CDC tabel Delta. Buat Stream dengan properti "read_mode"="cdc" untuk mengonsumsi data CDC secara inkremental.
Untuk dokumentasi lengkap Stream, lihat Objek Stream.
Sintaksis
CREATE STREAM [IF NOT EXISTS] <stream_name>
ON TABLE <delta_table_name> VERSION AS OF <v>
strmproperties ("read_mode"="cdc")Properti "read_mode"="cdc" mengatur Stream untuk mengonsumsi data CDC berdasarkan rentang versi kueri.
Contoh
-- Buat tabel sumber dengan CDC yang diaktifkan.
CREATE TABLE acid_with_cdc_stream (id1 BIGINT NOT NULL PRIMARY KEY, id2 BIGINT)
tblproperties ("transactional" = "true", "acid.cdc.mode.enable"="true");
-- Masukkan data.
INSERT INTO acid_with_cdc_stream VALUES (1, 1006), (2, 1008), (3, 1032);
-- Buat Stream mulai dari versi 1.
CREATE STREAM delta_table_stream ON TABLE acid_with_cdc_stream VERSION AS OF 1 strmproperties ("read_mode"="cdc");
-- Periksa Stream.
DESC STREAM delta_table_stream;Output:
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"}