La fonctionnalité Change Data Capture (CDC) enregistre les insertions, mises à jour et suppressions au niveau des lignes sur les tables Delta sous la forme d'un flux d'événements incrémentiel. Les systèmes en aval peuvent consommer ce flux pour le calcul incrémentiel, la synchronisation des données, le traitement de flux et l'audit, sans analyser l'intégralité de la table à chaque exécution.
La fonctionnalité CDC est actuellement en aperçu sur invitation. Pour obtenir des instructions d'utilisation, consultez la rubrique Présentation du calcul incrémentiel .
Cas d'utilisation
Calcul incrémentiel : consommez les enregistrements de modification pour mettre à jour les vues matérialisées sans retraiter l'intégralité de la table.
Traitement de flux : alimentez les tâches Flink ou d'autres processeurs de flux avec les enregistrements CDC.
Synchronisation des données multi-moteurs : propagez les modifications incrémentielles entre différents moteurs de calcul.
Audit des journaux : conservez un historique complet de toutes les opérations au niveau des lignes à des fins de conformité et de gouvernance.
Choisir un mode CDC
Le CDC prend en charge deux modes. Choisissez le mode adapté selon que votre charge de travail nécessite ou non des écritures en temps réel via Tunnel.
| |**CDC synchrone**
|
**CDC asynchrone**
| | --- | --- | --- | |
**Opérations SQL DML**
|
Prises en charge
|
Prises en charge
| |
**Écritures en temps réel via Tunnel**
|
Non prises en charge
|
Prises en charge
| |
**Disponibilité des données CDC**
|
Immédiatement après la fin de l'opération SQL
|
De manière asynchrone, après l'intervalle configuré
|
Activer le CDC sur une table Delta
CDC synchrone
Ajoutez les propriétés "acid.cdc.mode.enable"="true" et "acid.cdc.build.async" = "false" lors de la création d'une table 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: active la fonctionnalité CDC des tables Delta. Le mode asynchrone est utilisé par défaut. Les données CDC sont générées de manière synchrone lors des opérations SQL DML. Les écritures en temps réel via Tunnel ne sont pas prises en charge.acid.cdc.build.async: contrôle la méthode de génération des données CDC. Lorsqu'elle est définie sur « false », cette propriété crée une table CDC synchrone et les données CDC sont générées immédiatement après la fin des opérations SQL DML. Les écritures en temps réel via Tunnel ne sont pas prises en charge.cdc.data.retain.hours: propriété facultative. Période de rétention des données CDC, exprimée en heures. Plage : 1–168. Valeur par défaut : 24.Les tables pour lesquelles le CDC est activé ne prennent pas en charge l'opération INSERT INTO. Nous vous recommandons d'utiliser plutôt l'opération MERGE INTO.
Propriétés facultatives :
|**Propriété**
|
**Description**
|
**Valeur par défaut**
| | --- | --- | --- | |
`"cdc.data.retain.hours"="24"`
|
Période de rétention des données CDC, exprimée en heures. Plage : 1–168.
|
`24`
|
CDC asynchrone
Ajoutez les propriétés "acid.cdc.mode.enable"="true", "acid.cdc.build.async"="true" et "acid.cdc.build.interval"="300" lors de la création d'une table Delta. Vous pouvez également ajouter la propriété "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: active la fonctionnalité CDC des tables Delta. Le mode asynchrone est utilisé par défaut. Les données CDC sont générées de manière synchrone lors des opérations SQL DML. Les écritures en temps réel via Tunnel ne sont pas prises en charge.acid.cdc.build.async: active la génération asynchrone du CDC. Les écritures en temps réel dans la table via Tunnel sont prises en charge et les données CDC sont également générées de manière asynchrone pour les opérations SQL DML.acid.cdc.build.interval: intervalle de génération asynchrone, exprimé en secondes. Valeurs valides : [60-3540]. Configurez ce paramètre en fonction de vos scénarios métier ou incrémentiels.cdc.data.retain.hours: propriété facultative. Période de rétention des données CDC, exprimée en heures. Plage : 1–168. Valeur par défaut : 24.Autres paramètres facultatifs (niveau projet ou niveau session) :
odps.storage.orc.enable.memcmp.sort.key="true". Nous vous recommandons de l'activer au niveau du projet, ce qui améliore les performances de la génération asynchrone du CDC et des requêtes.
Interroger les données CDC
Utilisez la fonction table_changes pour interroger les enregistrements de modification CDC par version ou par plage d'horodatage.
Syntaxe
SELECT * FROM table_changes('<table_name>', <start> [, <end>]);Paramètres
|**Paramètre**
|
**Obligatoire**
|
**Type**
|
**Description**
| | --- | --- | --- | --- | |
`table_name`
|
Oui
|
—
|
Table Delta à interroger.
| |
`start`
|
Oui
|
BIGINT ou STRING
|
Version de départ pour l'interrogation des données CDC. Recherchez les numéros de version avec `SHOW HISTORY FOR TABLE
| |
`end`
|
Non
|
BIGINT ou STRING
|
Version de fin pour l'interrogation des données CDC. Si omis, la valeur par défaut correspond à la dernière version. Pour STRING, utilisez le format `yyyy-mm-dd hh:mi:ss`.
|
Colonnes renvoyées
En plus des colonnes de données de la table, la fonction table_changes renvoie trois colonnes système :
**Colonne**
|
**Description**
| | --- | --- | |
`__meta_timestamp`
|
Heure système à laquelle la ligne a été écrite.
| |
`__meta_op_type`
|
Type d'opération : `1` = INSERT, `0` = DELETE.
| |
`__meta_is_update`
|
Indique si la ligne fait partie d'une opération UPDATE : `1` = oui, `0` = non.
|
Combinez les colonnes __meta_op_type et __meta_is_update pour identifier l'opération complète :
**__meta_op_type**
|
**__meta_is_update**
|
**Opération**
| | --- | --- | --- | |
`1`
|
`0`
|
Nouvelle ligne issue d'une opération INSERT
| |
`1`
|
`1`
|
Valeur post-mise à jour (nouvelle valeur après l'opération UPDATE)
| |
`0`
|
`1`
|
Valeur pré-mise à jour (valeur d'origine avant l'opération UPDATE)
| |
`0`
|
`0`
|
Ligne supprimée
|
Exemple
Étape 1 : Créez une table avec le CDC activé.
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");Étape 2 : Insérez des données.
-- Data insertion time 2025-04-07 11:56:57
INSERT INTO acid_cdc_table VALUES ('1', '1006', 1006, 1006);
-- Data insertion time 2025-04-07 12:15:00
INSERT INTO acid_cdc_table VALUES ('1', '1008', 1008, 1008);
-- Data insertion time 2025-04-07 13:24:00
INSERT INTO acid_cdc_table VALUES ('1', '1032', 1032, 1032);
-- Data insertion time 2025-04-07 14:00:00
INSERT INTO acid_cdc_table VALUES ('1', '1045', 1045, 1045);
-- Data insertion time 2025-04-07 14:47:00
INSERT INTO acid_cdc_table VALUES ('1', '1045', 1045, 1045);Étape 3 : Recherchez les versions de la table.
SHOW HISTORY FOR TABLE acid_cdc_table;Résultat :
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Étape 4 : Interrogez les enregistrements CDC.
Interrogez toutes les modifications effectuées après 2025-04-07 12:00:00 (équivalent à la version 3) :
SELECT * FROM table_changes('acid_cdc_table', '2025-04-07 12:00:00');
-- Equivalent to
SELECT * FROM table_changes('acid_cdc_table', 3);Résultat :
+------------+------------+------------+------------+------------------+----------------+------------------+
| 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 |
+------------+------------+------------+------------+------------------+----------------+------------------+Interrogez les modifications dans une plage spécifique (de 2025-04-07 12:00:00 à 13:30:00, équivalent aux versions 3–6) :
SELECT * FROM table_changes('acid_cdc_table', '2025-04-07 12:00:00', '2025-04-07 13:30:00');
-- Equivalent to
SELECT * FROM table_changes('acid_cdc_table', 3, 6);Résultat :
+------------+------------+------------+------------+------------------+----------------+------------------+
| 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 |
+------------+------------+------------+------------+------------------+----------------+------------------+Consommer les données CDC avec un Stream
Un objet Stream suit un décalage de lecture dans l'historique CDC d'une table Delta. Créez un Stream avec la propriété "read_mode"="cdc" pour consommer les données CDC de manière incrémentielle.
Pour obtenir la documentation complète sur les objets Stream, consultez la rubrique Objet Stream.
Syntaxe
CREATE STREAM [IF NOT EXISTS] <stream_name>
ON TABLE <delta_table_name> VERSION AS OF <v>
strmproperties ("read_mode"="cdc")La propriété "read_mode"="cdc" configure le Stream pour qu'il consomme les données CDC en fonction de la plage de versions interrogée.
Exemple
-- Create the source table with CDC enabled.
CREATE TABLE acid_with_cdc_stream (id1 BIGINT NOT NULL PRIMARY KEY, id2 BIGINT)
tblproperties ("transactional" = "true", "acid.cdc.mode.enable"="true");
-- Insert data.
INSERT INTO acid_with_cdc_stream VALUES (1, 1006), (2, 1008), (3, 1032);
-- Create a Stream starting at version 1.
CREATE STREAM delta_table_stream ON TABLE acid_with_cdc_stream VERSION AS OF 1 strmproperties ("read_mode"="cdc");
-- Inspect the Stream.
DESC STREAM delta_table_stream;Résultat :
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"}