Realtime Compute for Apache Flink:Accéder à AnalyticDB for PostgreSQL

Limitations

• Realtime Compute for Apache Flink ne peut pas lire les données depuis AnalyticDB for PostgreSQL en mode serverless.

• Le connecteur AnalyticDB for PostgreSQL nécessite Ververica Runtime (VVR) version 6.0.0 ou ultérieure.

• AnalyticDB for PostgreSQL V7.0 nécessite VVR version 8.0.1 ou ultérieure.

Pour utiliser un connecteur personnalisé à la place, consultez la section Gérer les connecteurs personnalisés .

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

• Un espace de travail Flink entièrement géré. Consultez la section Activer Flink entièrement géré.

• Une instance AnalyticDB for PostgreSQL et un compte privilégié. Consultez les sections Créer une instance et Créer un compte privilégié.

• L'instance AnalyticDB for PostgreSQL et l'espace de travail Flink entièrement géré doivent se trouver dans le même cloud privé virtuel (VPC).

S'ils se trouvent dans des VPC différents, consultez la section Comment Flink entièrement géré accède-t-il à un service via plusieurs VPC ?

Étape 1 : Configurer une liste d'autorisation et préparer les données

1. Connectez-vous à la console AnalyticDB for PostgreSQL.

2. Ajoutez le bloc CIDR de l'espace de travail Flink entièrement géré à la liste d'autorisation de l'instance AnalyticDB for PostgreSQL.

   1. Identifiez le bloc CIDR du vSwitch utilisé par votre espace de travail Flink entièrement géré. Consultez la section Comment configurer une liste d'autorisation ?

   2. Ajoutez le bloc CIDR à la liste d'autorisation de l'instance AnalyticDB for PostgreSQL. Consultez la section Procédure.

   Si vous accédez à l'instance via Internet, ajoutez l'adresse IP publique à la place.

3. Sur la page des détails de l'instance, cliquez sur Log On to Database dans le coin supérieur droit, puis saisissez votre nom d'utilisateur et votre mot de passe. Pour plus de détails, consultez la section Utiliser des outils clients pour se connecter à une instance.

4. Créez une table de dimension nommée adbpg_dim_table et insérez 50 lignes de données d'exemple.

   -- Create the dimension table
   CREATE TABLE adbpg_dim_table(
     id int,
     username text,
     PRIMARY KEY(id)
   );
   
   -- Insert 50 rows: id ranges from 1 to 50, username is "username" followed by the row number
   INSERT INTO adbpg_dim_table(id, username)
   SELECT i, 'username'||i::text
   FROM generate_series(1, 50) AS t(i);

   Exécutez SELECT * FROM adbpg_dim_table ORDER BY id; pour vérifier les données insérées.

5. Créez une table de résultats nommée adbpg_sink_table afin que Flink puisse y écrire les données de sortie.

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

Étape 2 : Créer un brouillon de flux

1. Connectez-vous à la console Realtime Compute for Apache Flink, localisez votre espace de travail, puis cliquez sur Console dans la colonne Actions.

2. Dans le volet de navigation de gauche, accédez à Development > ETL. Dans le coin supérieur gauche de la page de l'éditeur SQL, cliquez sur New et sélectionnez New Blank Stream Draft.

3. Dans la boîte de dialogue New Draft, configurez les paramètres suivants.

   Parameter	Description	Example
   Name	Nom du brouillon. Il doit être unique au sein du projet.	adbpg-test
   Location	Dossier dans lequel le brouillon est enregistré. Cliquez sur l'icône à côté d'un dossier existant pour créer un sous-dossier.	Draft
   Engine Version	Version du moteur Flink. Consultez la section Versions du moteur pour obtenir des détails sur les versions et leur cycle de vie.	vvr-8.0.1-flink-1.17

4. Cliquez sur Create.

Étape 3 : Rédiger et déployer le brouillon

1. Copiez le code SQL suivant dans l'éditeur de code. Ce code définit trois tables et une jointure de recherche (lookup join) qui enrichit le flux Datagen avec les données utilisateur provenant d'AnalyticDB for PostgreSQL.

   -- Source table: Datagen generates sequential IDs (1-50) and random scores (70-100).
   -- No changes needed in the WITH clause for this example.
   CREATE TEMPORARY TABLE datagen_source (
     id INT,
     score INT
   ) WITH (
     'connector' = 'datagen',
     'fields.id.kind' = 'sequence',
     'fields.id.start' = '1',
     'fields.id.end' = '50',
     'fields.score.kind' = 'random',
     'fields.score.min' = '70',
     'fields.score.max' = '100'
   );
   
   -- Dimension table: backed by AnalyticDB for PostgreSQL.
   -- Flink queries this table at processing time to look up usernames by ID.
   -- Replace the WITH clause values with your actual connection details.
   CREATE TEMPORARY TABLE dim_adbpg(
     id int,
     username varchar,
     PRIMARY KEY(id) NOT ENFORCED
   ) WITH (
     'connector' = 'adbpg',
     'url' = 'jdbc:postgresql://gp-2ze****3tysk255b5-master.gpdb.rds.aliyuncs.com:5432/flinktest',
     'tablename' = 'adbpg_dim_table',
     'username' = 'flinktest',
     'password' = '${secret_values.adb_password}',
     'maxRetryTimes' = '2',
     'cache' = 'lru',
     'cacheSize' = '100'
   );
   
   -- Result table: Flink writes enriched records here.
   -- Replace the WITH clause values with your actual connection details.
   CREATE TEMPORARY TABLE sink_adbpg (
     id int,
     username varchar,
     score int
   ) WITH (
     'connector' = 'adbpg',
     'url' = 'jdbc:postgresql://gp-2ze****3tysk255b5-master.gpdb.rds.aliyuncs.com:5432/flinktest',
     'tablename' = 'adbpg_sink_table',
     'username' = 'flinktest',
     'password' = '${secret_values.adb_password}',
     'maxRetryTimes' = '2',
     'conflictMode' = 'ignore',
     'retryWaitTime' = '200'
   );
   
   -- Lookup join: for each record from datagen_source, Flink looks up the matching
   -- row in dim_adbpg at the time the record is processed (PROCTIME()).
   INSERT INTO sink_adbpg
   SELECT ts.id, ts.username, ds.score
   FROM datagen_source AS ds
   JOIN dim_adbpg FOR SYSTEM_TIME AS OF PROCTIME() AS ts
   ON ds.id = ts.id;

   À propos de la syntaxe de la jointure de recherche : FOR SYSTEM_TIME AS OF PROCTIME() indique à Flink d'interroger la table de dimension au moment où chaque enregistrement source est traité. Cela signifie que chaque enregistrement est enrichi avec les données de dimension existantes au moment du traitement. Les résultats déjà écrits ne sont pas mis à jour si la table de dimension change ultérieurement.

2. Mettez à jour les paramètres de connexion pour la table de dimension et la table de résultats. Remplacez les valeurs d'espace réservé dans les clauses WITH par vos propres détails de connexion AnalyticDB for PostgreSQL. La table source Datagen ne nécessite aucune modification. Pour la référence complète des paramètres et les mappages de types de données, consultez la section Connecteur AnalyticDB for PostgreSQL.

   Parameter	Required	Default	Description
   url	Oui	—	URL JDBC au format jdbc:postgresql://<Internal endpoint>:<Port>/<Database name>. Vous trouverez cette information sur la page Database Connection de l'instance dans la console AnalyticDB for PostgreSQL.
   tablename	Oui	—	Nom de la table dans la base de données AnalyticDB for PostgreSQL.
   username	Oui	—	Nom d'utilisateur pour accéder à la base de données.
   password	Oui	—	Mot de passe du compte de base de données.
   targetSchema	Non	public	Nom du schéma. Spécifiez ce paramètre uniquement si votre table ne se trouve pas dans le schéma public.
   maxRetryTimes	Non	—	Nombre maximal de tentatives après un échec d'écriture.
   cache	Non	—	Politique de cache pour les recherches dans la table de dimension. Définissez-la sur lru pour conserver en mémoire les entrées récemment consultées. La mise en cache LRU réduit le trafic vers la base de données et améliore le débit des recherches, mais les entrées mises en cache peuvent être obsolètes. Il s'agit d'un compromis entre le débit et l'actualité des données : ajustez cacheSize et tenez compte de votre tolérance aux données obsolètes avant d'activer cette option.
   cacheSize	Non	—	Nombre maximal d'entrées à mettre en cache. Des valeurs plus élevées réduisent les requêtes vers la base de données, mais consomment davantage de mémoire.
   conflictMode	Non	—	Action entreprise lorsqu'une écriture entre en conflit avec une clé primaire ou un index existant. Définissez ce paramètre sur ignore pour ignorer les lignes conflictuelles.
   retryWaitTime	Non	—	Temps d'attente en millisecondes entre les tentatives d'écriture.

3. Dans le coin supérieur droit de la page de l'éditeur SQL, cliquez sur Validate pour vérifier la syntaxe.

4. Cliquez sur Deploy.

5. Sur la page O&M > Deployments, localisez votre déploiement, puis cliquez sur Start dans la colonne Actions.

Étape 4 : Vérifier le résultat

1. Connectez-vous à la console AnalyticDB for PostgreSQL.

2. Cliquez sur Log On to Database. Pour plus de détails, consultez la section Se connecter à une instance depuis un client.

3. Exécutez la requête suivante pour afficher les enregistrements que Flink a écrits dans la table de résultats.

   SELECT * FROM adbpg_sink_table ORDER BY id;

   Le résultat doit contenir 50 lignes, chacune comprenant un ID utilisateur, le nom d'utilisateur correspondant issu de la table de dimension et un score aléatoire compris entre 70 et 100.

   La requête renvoie 7 enregistrements comportant trois colonnes : id, username et score. Les lignes 1 à 7 correspondent respectivement à username1 à username7, avec des scores de 94, 79, 70, 93, 71, 82 et 87.

Références

• Connecteur Datagen

• Connecteur AnalyticDB for PostgreSQL