O Change Data Capture (CDC) registra inserções, atualizações e exclusões no nível de linha em tabelas Delta como um fluxo de eventos incrementais. Sistemas downstream podem consumir esse fluxo para computação incremental, sincronização de dados, processamento de stream e auditoria, sem precisar varrer toda a tabela a cada execução.
O CDC está atualmente em visualização por convite. Para instruções de uso, consulte Incremental calculation overview .
Casos de uso
Computação incremental: Consuma registros de alteração para atualizar visualizações materializadas sem reprocessar a tabela inteira.
Processamento de stream: Alimente jobs do Flink ou outros processadores de stream com registros CDC.
Sincronização de dados entre múltiplos engines: Propague alterações incrementais entre diferentes engines de computação.
Auditoria de logs: Mantenha um registro completo de todas as operações no nível de linha para conformidade e governança.
Escolha um modo de CDC
O CDC oferece suporte a dois modos. Escolha com base na necessidade de escritas em tempo real via Tunnel na sua carga de trabalho.
|
CDC síncrono |
CDC assíncrono |
|
|
Operações SQL DML |
Suportado |
Suportado |
|
Escritas em tempo real via Tunnel |
Não suportado |
Suportado |
|
Disponibilidade dos dados CDC |
Imediatamente após a conclusão da operação SQL |
Assincronamente, após o intervalo configurado |
Ative o CDC em uma tabela Delta
CDC síncrono
Adicione as propriedades "acid.cdc.mode.enable"="true" e "acid.cdc.build.async" = "false" ao criar uma tabela 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: Após ativar o recurso CDC da Tabela Delta, ele é executado de forma assíncrona por padrão. O CDC é construído assincronamente quando alterações de dados são acionadas por operaçõesDML SQL. A geração de CDC também ocorre de forma assíncrona em cenários de escrita em tempo real via Tunnel.
acid.cdc.build.async: Controla como os dados CDC são construídos. Quando definido como "false", cria uma tabela CDC síncrona e gera dados CDC imediatamente após a conclusão das operações DML SQL. Escritas em tempo real pelo Tunnel não são suportadas.cdc.data.retain.hours: Propriedade opcional. Período de retenção para dados CDC, em horas. Intervalo: 1–168. Valor padrão: 24.Tabelas com CDC ativado não suportam a operação INSERT INTO. Recomendamos o uso de MERGE INTO como alternativa.
Propriedades opcionais:
|
Propriedade |
Descrição |
Padrão |
|
|
Período de retenção para dados CDC, em horas. Intervalo: 1–168. |
|
CDC assíncrono
Adicione "acid.cdc.mode.enable"="true", "acid.cdc.build.async"="true" e "acid.cdc.build.interval"="300" ao criar uma tabela Delta. Opcionalmente, adicione a propriedade "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: Após ativar o recurso CDC da Tabela Delta, ele é executado de forma assíncrona por padrão. O CDC é construído assincronamente quando alterações de dados são acionadas por operaçõesDML SQL. A geração de CDC também ocorre de forma assíncrona em cenários de escrita em tempo real via Tunnel.
acid.cdc.build.async: Habilita a construção assíncrona do CDC. Escritas em tempo real na tabela pelo Tunnel são suportadas, e os dados CDC também são gerados assincronamente para operações DML SQL.acid.cdc.build.interval: Intervalo para construção assíncrona, em segundos. Valores válidos: [60-3540]. Configure este parâmetro com base nos cenários de negócio ou incrementais.cdc.data.retain.hours: Propriedade opcional. Período de retenção para dados CDC, em horas. Intervalo: 1–168. Valor padrão: 24.Outros parâmetros opcionais (nível de projeto ou sessão):
odps.storage.orc.enable.memcmp.sort.key="true". Recomendamos ativar esta opção no nível do projeto, pois melhora o desempenho tanto da construção assíncrona do CDC quanto das consultas.
Consulte dados CDC
Use a função table_changes para consultar registros de alteração do CDC por versão ou intervalo de timestamp.
Sintaxe
SELECT * FROM table_changes('<table_name>', <start> [, <end>]);Parâmetros
|
Parâmetro |
Obrigatório |
Tipo |
Descrição |
|
|
Sim |
— |
A tabela Delta a ser consultada. |
|
|
Sim |
BIGINT ou STRING |
Versão inicial para a consulta de dados CDC. Encontre os números de versão com |
|
|
Não |
BIGINT ou STRING |
Versão final para a consulta de dados CDC. Assume a versão mais recente se omitido. Para STRING, use o formato |
Colunas retornadas
Além das colunas de dados da tabela, table_changes retorna três colunas de sistema:
|
Coluna |
Descrição |
|
|
Hora do sistema em que a linha foi escrita. |
|
|
Tipo de operação: |
|
|
Indica se a linha faz parte de um UPDATE: |
Combine __meta_op_type e __meta_is_update para identificar a operação completa:
|
__meta_op_type |
__meta_is_update |
Operação |
|
|
|
Nova linha proveniente de um INSERT |
|
|
|
Valor pós-atualização (o novo valor após o UPDATE) |
|
|
|
Valor pré-atualização (o valor original antes do UPDATE) |
|
|
|
Linha excluída |
Exemplo
Passo 1: Crie uma tabela com CDC ativado.
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");Passo 2: Insira dados.
-- 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);Passo 3: Encontre as versões da tabela.
SHOW HISTORY FOR TABLE acid_cdc_table;Saída:
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_COMPACTPasso 4: Consulte registros CDC.
Consulte todas as alterações após 2025-04-07 12:00:00 (equivalente à versão 3):
SELECT * FROM table_changes('acid_cdc_table', '2025-04-07 12:00:00');
-- Equivalent to
SELECT * FROM table_changes('acid_cdc_table', 3);Saída:
+------------+------------+------------+------------+------------------+----------------+------------------+
| 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 |
+------------+------------+------------+------------+------------------+----------------+------------------+Consulte alterações dentro de um intervalo específico (de 2025-04-07 12:00:00 até 13:30:00, equivalente às versões 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);Saída:
+------------+------------+------------+------------+------------------+----------------+------------------+
| 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 |
+------------+------------+------------+------------+------------------+----------------+------------------+Consuma dados CDC com um Stream
Um objeto Stream rastreia um offset de leitura no histórico CDC de uma tabela Delta. Crie um Stream com "read_mode"="cdc" para consumir dados CDC de forma incremental.
Para a documentação completa sobre Stream, consulte Stream object.
Sintaxe
CREATE STREAM [IF NOT EXISTS] <stream_name>
ON TABLE <delta_table_name> VERSION AS OF <v>
strmproperties ("read_mode"="cdc")A propriedade "read_mode"="cdc" configura o Stream para consumir dados CDC com base no intervalo de versões da consulta.
Exemplo
-- 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;Saída:
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"}