AnalyticDB:Utiliser Flink CDC pour s'abonner aux données complètes et incrémentielles en temps réel

Étape 1 : Préparer une table de test et des données de test

1. Connectez-vous à la console AnalyticDB for PostgreSQL et localisez l'instance cible. Cliquez sur l'ID de l'instance.

2. Dans le coin inférieur droit de la page Basic Information, cliquez sur Log On to Database.

3. Créez une base de données de test et une table source nommée adbpg_source_table. Insérez ensuite 50 lignes de données dans la table source.

   -- 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. Créez une table de destination nommée adbpg_sink_table pour permettre à Flink d'écrire les données résultantes.

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

Étape 2 : Créer un job Flink

1. Connectez-vous à la console Realtime Compute. Dans l'onglet Fully Managed Flink, localisez l'espace de travail cible et cliquez sur Console dans la colonne Actions.

2. Dans le volet de navigation de gauche, sélectionnez Development > ETL.

3. Dans la barre de menus supérieure, cliquez sur . Sélectionnez New Blank Stream Draft et configurez les paramètres ci-dessous.

   Paramètre du job	Description	Exemple
   Name	Nom du job. Remarque Le nom du job doit être unique au sein du projet actuel.	adbpg-test
   Location	Dossier dans lequel le fichier de code du job est stocké. Vous pouvez également cliquer sur l'icône à côté d'un dossier existant pour créer un sous-dossier.	Job Drafts
   Engine Version	Version du moteur Flink utilisée par le job. Pour plus d'informations sur les numéros de version du moteur, les mappages de versions et les jalons du cycle de vie, consultez Versions du moteur Flink.	vvr-6.0.7-flink-1.15

4. Cliquez sur Create.

Étape 3 : Écrire le code du job et déployer le job

1. Créez une source nommée datagen_source pour générer des données simulées et une source nommée source_adbpg pour capturer les modifications de données en temps réel depuis la base de données AnalyticDB for PostgreSQL. Joignez ensuite les deux sources et écrivez les résultats dans une table de destination nommée sink_adbpg. Les données traitées sont écrites dans AnalyticDB for PostgreSQL.

   Copiez le code de job suivant dans l'éditeur.

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

   Paramètres

   Paramètre	Obligatoire	Type de données	Description
   connector	Oui	STRING	Type de connecteur. Définissez la valeur sur adbpg-cdc pour la table source et sur adbpg pour la table de destination.
   hostname	Oui	STRING	Point de terminaison interne de l'instance AnalyticDB for PostgreSQL. Vous pouvez obtenir le point de terminaison interne sur la page Basic Information de l'instance.
   username	Oui	STRING	Compte de base de données et mot de passe de l'instance AnalyticDB for PostgreSQL.
   password	Oui	STRING
   database-name	Oui	STRING	Nom de la base de données.
   schema-name	Oui	STRING	Nom du schéma. Ce paramètre prend en charge les expressions régulières. Vous pouvez vous abonner à plusieurs schémas simultanément.
   table-name	Oui	STRING	Nom de la table. Ce paramètre prend en charge les expressions régulières. Vous pouvez vous abonner à plusieurs tables simultanément.
   port	Oui	INTEGER	Port d'AnalyticDB for PostgreSQL. La valeur est fixée à 5432.
   decoding.plugin.name	Oui	STRING	Nom du plug-in de décodage logique PostgreSQL. La valeur est fixée à pgoutput.
   slot.name	Oui	STRING	Nom du slot de décodage logique. Pour les tables source d'un même job Flink, utilisez la même valeur pour slot.name. Si différents jobs Flink impliquent la même table, définissez un slot.name unique pour chaque job. Cela permet d'éviter l'erreur suivante : PSQLException: ERROR: replication slot "debezium" is active for PID 974.
   debezium.*	Non	STRING	Contrôle le comportement du client Debezium avec une granularité plus fine. Par exemple, la définition de 'debezium.snapshot.mode' = 'never' désactive la fonctionnalité de snapshot. Pour plus d'informations, consultez les propriétés de configuration.
   scan.incremental.snapshot.enabled	Non	BOOLEAN	Indique si les snapshots incrémentiels sont activés. Valeurs possibles : false (par défaut) : les snapshots incrémentiels sont désactivés. true : les snapshots incrémentiels sont activés.
   scan.startup.mode	Non	STRING	Mode de démarrage pour la consommation des données. Valeurs possibles : initial (par défaut) : lors du premier démarrage du job, celui-ci analyse toutes les données historiques, puis lit les dernières données WAL (Write-Ahead Logging). Cela assure une transition transparente entre les données complètes et incrémentielles. latest-offset : lors du premier démarrage du job, celui-ci n'analyse pas les données historiques. Il commence la lecture à partir de la fin du WAL, c'est-à-dire la dernière position de journal. Il capture uniquement les modifications de données survenant après le démarrage du connecteur. snapshot : analyse toutes les données historiques et lit les nouvelles entrées WAL générées pendant l'analyse complète. Le job s'arrête une fois l'analyse complète terminée.
   changelog-mode	Non	STRING	Mode de journal des modifications pour l'encodage des changements de flux. Valeurs possibles : ALL (par défaut) : prend en charge tous les types d'opérations, y compris INSERT, DELETE, UPDATE_BEFORE et UPDATE_AFTER. UPSERT : prend en charge uniquement les opérations UPSERT, y compris INSERT, DELETE et UPDATE_AFTER.
   heartbeat.interval.ms	Non	DURATION	Intervalle d'envoi des paquets heartbeat. La valeur par défaut est de 30 secondes. L'unité est la milliseconde. Le connecteur CDC AnalyticDB for PostgreSQL envoie des paquets heartbeat à la base de données pour garantir que le décalage du slot avance continuellement. Si les données de la table ne changent pas fréquemment, définissez ce paramètre sur une valeur raisonnable pour nettoyer rapidement les journaux WAL et éviter de gaspiller de l'espace disque.
   scan.incremental.snapshot.chunk.key-column	Non	STRING	Spécifie une colonne à utiliser pour le découpage en chunks pendant la phase de snapshot. Par défaut, la première colonne de la clé primaire est sélectionnée.
   url	Oui	STRING	Le format est jdbc:postgresql://<Address>:<PortId>/<DatabaseName>.

2. En haut de la page de développement du job, cliquez sur Deep Check pour effectuer une vérification syntaxique.

3. Cliquez sur Deploy, puis sur Confirm.

4. Dans le coin supérieur droit, cliquez sur Operations. Sur la page Deployments, cliquez sur Start.

Étape 4 : Afficher les données écrites par Flink

1. Exécutez les instructions suivantes dans la base de données de test pour afficher les données écrites par Flink.

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

2. Insérez 50 lignes de données supplémentaires dans la table source. Vérifiez ensuite le nombre total de lignes de données incrémentielles que Flink écrit dans la table de destination.

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

   Le résultat est indiqué ci-dessous.

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

Étape 1 : Préparer les tables de test

Dans l'instance AnalyticDB for PostgreSQL, créez deux tables source.

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

Étape 2 : Préparer les ressources Kafka

1. Achetez et déployez une instance Kafka.

2. Ajoutez le bloc CIDR de l'espace de travail Flink à la liste d'autorisation de l'instance Kafka.

3. Créez des ressources dans l'instance Kafka.

Étape 3 : Créer un job Flink

1. Connectez-vous à la console Realtime Compute. Dans l'onglet Fully Managed Flink, localisez l'espace de travail cible et cliquez sur Console dans la colonne Actions.

2. Dans le volet de navigation de gauche, sélectionnez Development > ETL.

3. Dans la barre de menus supérieure, cliquez sur . Sélectionnez New Blank Stream Draft et configurez les paramètres du job.

4. Cliquez sur Create.

Étape 4 : Écrire le code du job et déployer le job

1. Écrivez un job SQL dans l'espace de travail Flink. Copiez le code de job suivant dans l'éditeur et remplacez les configurations par vos valeurs réelles.

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

   Prenez note des points suivants concernant ce job SQL :

   • Pour les tâches de synchronisation multi-tables, nous vous recommandons d'utiliser une table source unique pour capturer les données de plusieurs tables, comme illustré dans cet exemple SQL. Vous devez définir toutes les colonnes de toutes les tables source dans cette table source. Si des noms de colonnes sont dupliqués, n'en conservez qu'un seul. Lors de l'écriture dans la table de destination, utilisez les métadonnées METADATA table_name pour router les données vers la table spécifiée. Cette approche nécessite la création d'un seul slot de réplication dans AnalyticDB for PostgreSQL. Cela réduit l'utilisation des ressources de la base de données source, améliore les performances de synchronisation et simplifie la maintenance future.

   • Utilisez le paramètre table-name pour spécifier plusieurs tables source. Placez les noms de table entre parenthèses et séparez-les par des barres verticales (|), par exemple (table1|table2|table3).

   • La définition de debezium.snapshot.mode sur never signifie que seules les données incrémentielles de la table source sont synchronisées. Pour synchroniser à la fois les données complètes et incrémentielles, modifiez le paramètre pour le définir sur initial.

2. En haut de la page de développement du job, cliquez sur Deep Check pour effectuer une vérification syntaxique.

3. Cliquez sur Deploy, puis sur OK.

4. Dans le coin supérieur droit, cliquez sur Operations. Sur la page Deployments, cliquez sur Start.

Étape 5 : Insérer des données de test

Dans l'instance AnalyticDB for PostgreSQL, mettez à jour les données des deux tables source et observez les modifications des messages dans la rubrique Kafka.

Vous pouvez utiliser l'instruction SQL suivante pour insérer des données de test :

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