AnalyticDB:Use Flink CDC to subscribe to full and incremental data in real time (invitational preview)

Etapa 1: Preparar uma tabela de teste e dados de teste

1. Faça logon no console do AnalyticDB for PostgreSQL e localize a instância desejada. Clique em ID da instância.

2. No canto inferior direito da página Basic Information, clique em Log On to Database.

3. Crie um banco de dados de teste e uma tabela de origem chamada adbpg_source_table. Em seguida, insira 50 linhas de dados nessa tabela.

   -- Create a test database.
   CREATE DATABASE testdb;
   -- Switch to the testdb database and create a schema.
   CREATE SCHEMA testschema;
   -- Create a source table named adbpg_source_table.
   CREATE TABLE testschema.adbpg_source_table(
     id int,
     username text,
     PRIMARY KEY(id)
   );
   -- Insert 50 rows of data into the adbpg_source_table table.
   INSERT INTO testschema.adbpg_source_table(id, username)
   SELECT i, 'username'||i::text
   FROM generate_series(1, 50) AS t(i);

4. Crie uma tabela de destino chamada adbpg_sink_table para que o Flink grave os dados resultantes.

   CREATE TABLE testschema.adbpg_sink_table(
     id int,
     username text,
     score int
   );

Etapa 2: Criar um job do Flink

1. Faça logon no console do Realtime Compute. Na aba Fully Managed Flink, localize o workspace desejado e clique em Console na coluna Actions.

2. No painel de navegação à esquerda, escolha Development > ETL.

3. Na barra de menu superior, clique em . Selecione New Blank Stream Draft e configure os parâmetros abaixo.

   Parâmetro do job	Descrição	Exemplo
   Name	Nome do job. Nota O nome do job deve ser exclusivo dentro do projeto atual.	adbpg-test
   Location	Pasta onde o arquivo de código do job será armazenado. Também é possível clicar em ícone ao lado de uma pasta existente para criar uma subpasta.	Job Drafts
   Engine Version	Versão do mecanismo Flink utilizada pelo job. Para mais informações sobre números de versão, mapeamentos de versões e marcos do ciclo de vida, consulte Flink engine versions.	vvr-6.0.7-flink-1.15

4. Clique em Create.

Etapa 3: Escrever o código do job e implantá-lo

1. Crie uma origem chamada datagen_source para gerar dados simulados e outra origem chamada source_adbpg para capturar alterações de dados em tempo real do banco de dados AnalyticDB for PostgreSQL. Em seguida, faça o join das duas origens e grave os resultados em uma tabela de destino chamada sink_adbpg. Os dados processados serão gravados no AnalyticDB for PostgreSQL.

   Copie o seguinte código do job para o editor.

   ---Create a Datagen source table to generate streaming data using the Datagen connector.
   CREATE TEMPORARY TABLE datagen_source (
    id INT,
    score INT
   ) WITH (
    'connector' = 'datagen',
    'fields.id.kind'='sequence',
    'fields.id.start'='1',
    'fields.id.end'='100',
    'fields.score.kind'='random',
    'fields.score.min'='70',
    'fields.score.max'='100'
   );
   
   --Create an adbpg source table to capture data changes of the adbpg_source_table table based on slot.name and pgoutput using the adbpg-cdc connector.
   CREATE TEMPORARY TABLE source_adbpg(
    id int,
    username varchar,
    PRIMARY KEY(id) NOT ENFORCED
   ) WITH(
     'connector' = 'adbpg-cdc', 
     'hostname' = 'gp-bp16v8cgx46ns****-master.gpdb.rds.aliyuncs.com',
     'port' = '5432',
     'username' = 'account****',
     'password' = 'password****',
     'database-name' = 'testdb',
     'schema-name' = 'testschema',
     'table-name' = 'adbpg_source_table',
     'slot.name' = 'flink',
     'decoding.plugin.name' = 'pgoutput'
   );
   
   --Create an adbpg sink table to write the processed results to the destination table adbpg_sink_table in the database.
   CREATE TEMPORARY TABLE sink_adbpg (
     id int,
     username varchar,
     score int
   ) WITH (
     'connector' = 'adbpg', 
     'url' = 'jdbc:postgresql://gp-bp16v8cgx46ns****-master.gpdb.rds.aliyuncs.com:5432/testdb',
     'tablename' = 'testschema.adbpg_sink_table',  
     'username' = 'account****',
     'password' = 'password****',
     'maxRetryTimes' = '2',
     'batchsize' = '5000',
     'conflictMode' = 'ignore',
     'writeMode' = 'insert',
     'retryWaitTime' = '200'
   );
   
   -- Write the join results of the datagen_source and source_adbpg tables to the adbpg sink table.
   INSERT INTO sink_adbpg
   SELECT ts.id,ts.username,ds.score FROM datagen_source AS ds
   JOIN source_adbpg AS ts ON ds.id = ts.id;

   Parâmetros

   Parâmetro	Obrigatório	Tipo de dado	Descrição
   connector	Sim	STRING	Tipo do conector. Defina o valor como adbpg-cdc para a tabela de origem e adbpg para a tabela de destino.
   hostname	Sim	STRING	Endpoint interno da instância do AnalyticDB for PostgreSQL. Obtenha o internal endpoint na página Basic Information da instância.
   username	Sim	STRING	Conta e senha do banco de dados da instância do AnalyticDB for PostgreSQL.
   password	Sim	STRING
   database-name	Sim	STRING	Nome do banco de dados.
   schema-name	Sim	STRING	Nome do schema. Este parâmetro aceita expressões regulares, permitindo a assinatura de múltiplos schemas simultaneamente.
   table-name	Sim	STRING	Nome da tabela. Suporta expressões regulares para assinar várias tabelas ao mesmo tempo.
   port	Sim	INTEGER	Porta do AnalyticDB for PostgreSQL. O valor é fixo em 5432.
   decoding.plugin.name	Sim	STRING	Nome do plug-in de decodificação lógica do PostgreSQL. Valor fixo: pgoutput.
   slot.name	Sim	STRING	Nome do slot de decodificação lógica. Para tabelas de origem no mesmo job do Flink, utilize o mesmo valor para slot.name. Se jobs diferentes do Flink envolverem a mesma tabela, defina um slot.name exclusivo para cada job. Isso evita o erro: PSQLException: ERROR: replication slot "debezium" is active for PID 974.
   debezium.*	Não	STRING	Controla o comportamento do cliente Debezium com maior granularidade. Por exemplo, definir 'debezium.snapshot.mode' = 'never' desativa o recurso de snapshot. Para mais detalhes, consulte as propriedades de configuração.
   scan.incremental.snapshot.enabled	Não	BOOLEAN	Define se snapshots incrementais devem ser ativados. Valores válidos: false (padrão): Snapshots incrementais desativados. true: Snapshots incrementais ativados.
   scan.startup.mode	Não	STRING	Modo de inicialização para consumo de dados. Valores válidos: initial (padrão): Na primeira execução do job, todos os dados históricos são verificados antes da leitura dos dados mais recentes do WAL (write-ahead logging). Isso proporciona uma transição contínua entre dados completos e incrementais. latest-offset: O job ignora dados históricos na primeira execução e começa a ler a partir do final do WAL (posição mais recente do log). Captura apenas alterações ocorridas após o início do conector. snapshot: Verifica todos os dados históricos e lê novas entradas WAL geradas durante essa varredura completa. O job é encerrado ao concluir a varredura total.
   changelog-mode	Não	STRING	Modo de changelog para codificar alterações no fluxo. Valores válidos: ALL (padrão): Suporta todos os tipos de operação, incluindo INSERT, DELETE, UPDATE_BEFORE e UPDATE_AFTER. UPSERT: Suporta apenas operações UPSERT, abrangendo INSERT, DELETE e UPDATE_AFTER.
   heartbeat.interval.ms	Não	DURATION	Intervalo para envio de pacotes de heartbeat. O valor padrão é 30 segundos, especificado em milissegundos. O conector CDC do AnalyticDB for PostgreSQL envia heartbeats ao banco de dados para garantir o avanço contínuo do offset do slot. Se os dados da tabela não mudarem com frequência, configure este parâmetro adequadamente para limpar logs WAL prontamente e evitar desperdício de espaço em disco.
   scan.incremental.snapshot.chunk.key-column	Não	STRING	Especifique uma coluna para particionamento (chunking) durante a fase de snapshot. Por padrão, a primeira coluna da chave primária é selecionada.
   url	Sim	STRING	Formato: jdbc:postgresql://<Address>:<PortId>/<DatabaseName>.

2. No topo da página de desenvolvimento do job, clique em Deep Check para validar a sintaxe.

3. Clique em Deploy e, em seguida, clique em Confirm.

4. No canto superior direito, clique em Operations. Na página Deployments, clique em Start.

Etapa 4: Visualizar os dados gravados pelo Flink

1. Execute as instruções abaixo no banco de dados de teste para visualizar os dados gravados pelo Flink.

   SELECT * FROM testschema.adbpg_sink_table;
   SELECT COUNT(*) FROM testschema.adbpg_sink_table; 

2. Insira mais 50 linhas de dados na tabela de origem. Depois, verifique o total de linhas incrementais que o Flink gravou na tabela de destino.

   -- Insert 50 rows of incremental data into the source table.
   INSERT INTO testschema.adbpg_source_table(id, username)
   SELECT i, 'username'||i::text
   FROM generate_series(51, 100) AS t(i);
   
   -- Check the new data in the destination table.
   SELECT COUNT(*) FROM testschema.adbpg_sink_table where id > 50;

   O resultado é exibido abaixo.

    count 
   -------
       50
   (1 row)

Etapa 1: Preparar tabelas de teste

Na instância do AnalyticDB for PostgreSQL, crie duas tabelas de origem.

CREATE TABLE products (
    product_id SERIAL PRIMARY KEY,
    product_name VARCHAR(200) NOT NULL,
    sku CHAR(12) NOT NULL,
    description TEXT,
    price NUMERIC(10,2) NOT NULL,
    discount_price DECIMAL(10,2),
    stock_quantity INTEGER DEFAULT 0,
    weight REAL,
    volume DOUBLE PRECISION,
    dimensions BOX,
    release_date DATE,
    is_featured BOOLEAN DEFAULT FALSE,
    rating FLOAT,
    warranty_period INTERVAL,
    metadata JSON,
    tags TEXT[]
);

CREATE TABLE documents (
    document_id UUID PRIMARY KEY,
    title VARCHAR(200) NOT NULL,
    content TEXT,
    summary TEXT,
    publication_date TIMESTAMP WITHOUT TIME ZONE,
    last_updated TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    author_id BIGINT,
    file_data BYTEA,
    xml_content XML,
    json_metadata JSON,
    reading_time INTERVAL,
    is_public BOOLEAN DEFAULT TRUE,
    views_count INTEGER DEFAULT 0,
    category VARCHAR(50),
    tags TEXT[]
);

Etapa 2: Preparar recursos do Kafka

1. Purchase and deploy a Kafka instance.

2. Adicione o bloco CIDR do workspace do Flink à lista de permissões da instância do Kafka.

3. Create resources in the Kafka instance.

Etapa 3: Criar um job do Flink

1. Faça logon no console do Realtime Compute. Na aba Fully Managed Flink, localize o workspace desejado e clique em Console na coluna Actions.

2. No painel de navegação à esquerda, escolha Development > ETL.

3. Na barra de menu superior, clique em . Selecione New Blank Stream Draft e configure os parâmetros do job.

4. Clique em Create.

Etapa 4: Escrever o código do job e implantá-lo

1. Escreva um job SQL no workspace do Flink. Copie o código do job abaixo para o editor e substitua as configurações pelos seus valores reais.

   -- Use one source to capture data from multiple tables
   CREATE TEMPORARY TABLE ADBPGSource(
       table_name STRING METADATA FROM 'table_name' VIRTUAL,
       row_kind STRING METADATA FROM 'row_kind' VIRTUAL,
       product_id BIGINT,
       product_name STRING,
       sku STRING,
       description STRING,
       price STRING,
       discount_price STRING,
       stock_quantity INT,
       weight STRING,
       volume STRING,
       dimensions STRING,
       release_date STRING,
       is_featured BOOLEAN,
       rating FLOAT,
       warranty_period STRING,
       metadata STRING,
       tags STRING,
       document_id STRING,
       title STRING,
       content STRING,
       summary STRING,
       publication_date STRING,
       last_updated STRING,
       author_id BIGINT,
       file_data STRING,
       xml_content STRING,
       json_metadata STRING,
       reading_time STRING,
       is_public BOOLEAN,
       views_count INT,
       category STRING
   ) WITH (
     'connector' = 'adbpg-cdc',
     'hostname' = 'gp-2zev887z58390***-master.gpdb.rds.aliyuncs.com',
     'port' = '5432',
     'username' = 'account****',
     'password' = 'password****',
     'database-name' = 'testdb',
     'schema-name' = 'public',
     'table-name' = '(products|documents)',
     'slot.name' = 'flink',
     'decoding.plugin.name' = 'pgoutput',
     'debezium.snapshot.mode' = 'never'
   );
   
   CREATE TEMPORARY TABLE KafkaProducts (
       product_id BIGINT,
       product_name STRING,
       sku STRING,
       description STRING,
       price STRING,
       discount_price STRING,
       stock_quantity INT,
       weight STRING,
       volume STRING,
       dimensions STRING,
       release_date STRING,
       is_featured BOOLEAN,
       rating FLOAT,
       warranty_period STRING,
       metadata STRING,
       tags STRING,
       PRIMARY KEY(product_id) NOT ENFORCED
   ) WITH (
     'connector' = 'upsert-kafka',
     'topic' = '****',
     'properties.bootstrap.servers' = 'alikafka-post-cn-****-1-vpc.alikafka.aliyuncs.com:9092',
     'key.format'='avro',
     'value.format'='avro'
   );
   
   CREATE TEMPORARY TABLE KafkaDocuments (
       document_id STRING,
       title STRING,
       content STRING,
       summary STRING,
       publication_date STRING,
       last_updated STRING,
       author_id BIGINT,
       file_data STRING,
       xml_content STRING,
       json_metadata STRING,
       reading_time STRING,
       is_public BOOLEAN,
       views_count INT,
       category STRING,
       tags STRING,
       PRIMARY KEY(document_id) NOT ENFORCED
   ) WITH (
     'connector' = 'upsert-kafka',
     'topic' = '****',
     'properties.bootstrap.servers' = 'alikafka-post-cn-****-1-vpc.alikafka.aliyuncs.com:9092',
     'key.format'='avro',
     'value.format'='avro'
   );
   
   -- Use a STATEMENT SET to wrap multiple statements
   BEGIN STATEMENT SET;
   -- Use the table_name METADATA to route data to the destination table
   INSERT INTO KafkaProducts
   SELECT product_id,product_name,sku,description,price,discount_price,stock_quantity,weight,volume,dimensions,release_date,is_featured,rating,warranty_period,metadata,tags
   FROM ADBPGSource
   WHERE table_name = 'products';
   
   INSERT INTO KafkaDocuments
   SELECT document_id,title,content,summary,publication_date,last_updated,author_id,file_data,xml_content,json_metadata,reading_time,is_public,views_count,category,tags
   FROM ADBPGSource
   WHERE table_name = 'documents';
   
   END;

   Observe os seguintes pontos sobre este job SQL:

   • Para tarefas de sincronização de múltiplas tabelas, recomenda-se usar uma única tabela de origem para capturar dados de várias tabelas, conforme demonstrado neste exemplo SQL. Todas as colunas de todas as tabelas de origem devem ser definidas nesta tabela única. Se houver nomes de colunas duplicados, mantenha apenas um. Ao gravar na tabela de destino, utilize o METADATA table_name para rotear os dados para a tabela especificada. Essa abordagem exige a criação de apenas um Replication Slot no AnalyticDB for PostgreSQL, reduzindo o uso de recursos do banco de dados de origem, melhorando o desempenho da sincronização e simplificando a manutenção futura.

   • Utilize o parâmetro table-name para especificar múltiplas tabelas de origem. Coloque os nomes das tabelas entre parênteses e separe-os por barras verticais (|), por exemplo: (table1|table2|table3).

   • Definir debezium.snapshot.mode como never significa que apenas dados incrementais da tabela de origem serão sincronizados. Para sincronizar dados completos e incrementais, altere a configuração para initial.

2. No topo da página de desenvolvimento do job, clique em Deep Check para validar a sintaxe.

3. Clique em Deploy e, em seguida, clique em OK.

4. No canto superior direito, clique em Operations. Na página Deployments, clique em Start.

Etapa 5: Inserir dados de teste

Na instância do AnalyticDB for PostgreSQL, atualize os dados nas duas tabelas de origem e observe the message changes no tópico do Kafka.

Utilize a instrução SQL abaixo para inserir dados de teste:

INSERT INTO products (
    product_name, sku, description, price, discount_price, stock_quantity, weight, volume, dimensions, release_date, is_featured, rating, warranty_period, metadata, tags
) VALUES (
    'Test Product', 'Test-2025', 'A piece of test product data', 299.99, 279.99, 150, 50.5, 120.75, '(10,20),(30,40)', '2023-05-01', TRUE, 4.8, INTERVAL '1 year', '{"brand": "TechCo", "model": "X1"}', '{"Test1", "Test2"}'
);