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

Paso 1: Preparar una tabla de prueba y datos de prueba

1. Inicie sesión en la consola de AnalyticDB for PostgreSQL y localice la instancia objetivo. Haga clic en el ID de la instancia.

2. En la esquina inferior derecha de la página Basic Information, haga clic en Log On to Database.

3. Cree una base de datos de prueba y una tabla de origen llamada adbpg_source_table. A continuación, inserte 50 filas de datos en la tabla de origen.

   -- 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. Cree una tabla de destino llamada adbpg_sink_table para que Flink escriba los datos resultantes.

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

Paso 2: Crear un trabajo de Flink

1. Inicie sesión en la consola de Realtime Compute. En la pestaña Fully Managed Flink, busque el espacio de trabajo deseado y haga clic en Console en la columna Actions.

2. En el panel de navegación izquierdo, seleccione Development > ETL.

3. En la barra de menú superior, haga clic en . Seleccione New Blank Stream Draft y configure los siguientes parámetros.

   Parámetro del trabajo	Descripción	Ejemplo
   Name	Nombre del trabajo. Nota El nombre del trabajo debe ser único dentro del proyecto actual.	adbpg-test
   Location	Carpeta donde se almacena el archivo de código del trabajo. También puede hacer clic en el icono junto a una carpeta existente para crear una subcarpeta.	Job Drafts
   Engine Version	Versión del motor de Flink que utiliza el trabajo. Para obtener más información sobre números de versión del motor, asignaciones de versiones e hitos del ciclo de vida, consulte Flink engine versions.	vvr-6.0.7-flink-1.15

4. Haga clic en Create.

Paso 3: Escribir el código del trabajo y desplegarlo

1. Cree un origen llamado datagen_source para generar datos simulados y otro origen llamado source_adbpg para capturar cambios de datos en tiempo real desde la base de datos de AnalyticDB for PostgreSQL. Luego, combine ambos orígenes y escriba los resultados en una tabla de destino denominada sink_adbpg. Los datos procesados se escribirán en AnalyticDB for PostgreSQL.

   Copie el siguiente código de trabajo en el 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	Obligatorio	Tipo de dato	Descripción
   connector	Sí	STRING	Tipo de conector. Establezca el valor en adbpg-cdc para la tabla de origen y en adbpg para la tabla de destino.
   hostname	Sí	STRING	Endpoint interno de la instancia de AnalyticDB for PostgreSQL. Puede obtener el internal endpoint en la página Basic Information de la instancia.
   username	Sí	STRING	Cuenta y contraseña de la base de datos de la instancia de AnalyticDB for PostgreSQL.
   password	Sí	STRING
   database-name	Sí	STRING	Nombre de la base de datos.
   schema-name	Sí	STRING	Nombre del esquema. Este parámetro admite expresiones regulares. Puede suscribirse a varios esquemas simultáneamente.
   table-name	Sí	STRING	Nombre de la tabla. Este parámetro admite expresiones regulares. Puede suscribirse a varias tablas simultáneamente.
   port	Sí	INTEGER	Puerto de AnalyticDB for PostgreSQL. El valor fijo es 5432.
   decoding.plugin.name	Sí	STRING	Nombre del plugin de decodificación lógica de PostgreSQL. El valor fijo es pgoutput.
   slot.name	Sí	STRING	Nombre del slot de decodificación lógica. Para las tablas de origen dentro del mismo trabajo de Flink, utilice el mismo valor para slot.name. Si distintos trabajos de Flink involucran la misma tabla, establezca un slot.name único para cada trabajo. Esto evita el siguiente error: PSQLException: ERROR: replication slot "debezium" is active for PID 974.
   debezium.*	No	STRING	Controla el comportamiento del cliente Debezium con mayor granularidad. Por ejemplo, establecer 'debezium.snapshot.mode' = 'never' deshabilita la función de instantánea. Para obtener más información, consulte las propiedades de configuración.
   scan.incremental.snapshot.enabled	No	BOOLEAN	Especifica si se habilitan las instantáneas incrementales. Valores válidos: false (predeterminado): Las instantáneas incrementales están deshabilitadas. true: Las instantáneas incrementales están habilitadas.
   scan.startup.mode	No	STRING	Modo de inicio para el consumo de datos. Valores válidos: initial (predeterminado): Cuando el trabajo se inicia por primera vez, escanea todos los datos históricos y luego lee los datos más recientes del registro de escritura anticipada (WAL). Esto proporciona una transición fluida entre datos completos e incrementales. latest-offset: Cuando el trabajo se inicia por primera vez, no escanea datos históricos. Comienza a leer desde el final del WAL, que corresponde a la posición más reciente del registro. Solo captura los cambios de datos que ocurren después de que se inicia el conector. snapshot: Escanea todos los datos históricos y lee las nuevas entradas WAL generadas durante el escaneo completo. El trabajo se detiene una vez finalizado el escaneo completo.
   changelog-mode	No	STRING	Modo de registro de cambios para codificar modificaciones en el flujo. Valores válidos: ALL (predeterminado): Admite todos los tipos de operación, incluidos INSERT, DELETE, UPDATE_BEFORE y UPDATE_AFTER. UPSERT: Solo admite operaciones UPSERT, incluidas INSERT, DELETE y UPDATE_AFTER.
   heartbeat.interval.ms	No	DURATION	Intervalo para enviar paquetes de latido. El valor predeterminado es 30 segundos. La unidad es milisegundos. El conector CDC de AnalyticDB for PostgreSQL envía paquetes de latido a la base de datos para garantizar que el desplazamiento del slot avance continuamente. Si los datos de la tabla no cambian con frecuencia, establezca este parámetro en un valor razonable para limpiar oportunamente los registros WAL y evitar el desperdicio de espacio en disco.
   scan.incremental.snapshot.chunk.key-column	No	STRING	Especifica una columna para dividir en fragmentos durante la fase de instantánea. De forma predeterminada, se selecciona la primera columna de la clave primaria.
   url	Sí	STRING	El formato es jdbc:postgresql://<Address>:<PortId>/<DatabaseName>.

2. En la parte superior de la página de desarrollo de trabajos, haga clic en Deep Check para realizar una comprobación de sintaxis.

3. Haga clic en Deploy y luego en Confirm.

4. En la esquina superior derecha, haga clic en Operations. En la página Deployments, haga clic en Start.

Paso 4: Consultar los datos escritos por Flink

1. Ejecute las siguientes instrucciones en la base de datos de prueba para consultar los datos escritos por Flink.

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

2. Inserte 50 filas adicionales de datos en la tabla de origen. Después, compruebe el número total de filas de datos incrementales que Flink escribe en la tabla 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;

   El resultado se muestra a continuación.

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

Paso 1: Preparar tablas de prueba

En la instancia de AnalyticDB for PostgreSQL, cree dos tablas de origen.

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[]
);

Paso 2: Preparar recursos de Kafka

1. Purchase and deploy a Kafka instance.

2. Agregue el bloque CIDR del espacio de trabajo de Flink a la lista de permitidos de la instancia de Kafka.

3. Create resources in the Kafka instance.

Paso 3: Crear un trabajo de Flink

1. Inicie sesión en la consola de Realtime Compute. En la pestaña Fully Managed Flink, busque el espacio de trabajo deseado y haga clic en Console en la columna Actions.

2. En el panel de navegación izquierdo, seleccione Development > ETL.

3. En la barra de menú superior, haga clic en . Seleccione New Blank Stream Draft y configure los parámetros del trabajo.

4. Haga clic en Create.

Paso 4: Escribir el código del trabajo y desplegarlo

1. Escriba un trabajo SQL en el espacio de trabajo de Flink. Copie el siguiente código de trabajo en el editor y reemplace las configuraciones con sus valores reales.

   -- 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;

   Tome en cuenta los siguientes puntos sobre este trabajo SQL:

   • Para tareas de sincronización de múltiples tablas, se recomienda utilizar una sola tabla de origen para capturar datos de varias tablas, como se muestra en este ejemplo SQL. Debe definir todas las columnas de todas las tablas de origen en esta única tabla de origen. Si hay nombres de columna duplicados, conserve solo uno. Al escribir en la tabla de destino, utilice METADATA table_name para enrutar los datos a la tabla especificada. Este enfoque requiere crear únicamente un Replication Slot en AnalyticDB for PostgreSQL, lo cual reduce el uso de recursos de la base de datos de origen, mejora el rendimiento de la sincronización y simplifica el mantenimiento futuro.

   • Utilice el parámetro table-name para especificar varias tablas de origen. Encierre los nombres de las tablas entre paréntesis y sepárelos con barras verticales (|), por ejemplo: (table1|table2|table3).

   • Establecer debezium.snapshot.mode en never significa que solo se sincronizan los datos incrementales de la tabla de origen. Para sincronizar tanto datos completos como incrementales, cambie la configuración a initial.

2. En la parte superior de la página de desarrollo de trabajos, haga clic en Deep Check para realizar una comprobación de sintaxis.

3. Haga clic en Deploy y luego en OK.

4. En la esquina superior derecha, haga clic en Operations. En la página Deployments, haga clic en Start.

Paso 5: Insertar datos de prueba

En la instancia de AnalyticDB for PostgreSQL, actualice los datos en las dos tablas de origen y observe the message changes en el tema de Kafka.

Puede utilizar la siguiente instrucción SQL para insertar datos de prueba:

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"}'
);