Écrire des données dans AnalyticDB for PostgreSQL

Limites d'utilisation

Cette fonctionnalité ne prend pas en charge le mode Serverless d'AnalyticDB for PostgreSQL.
Seul Realtime Compute for Apache Flink VVR 6.0.0 ou une version ultérieure prend en charge le connecteur AnalyticDB for PostgreSQL.
Seul Realtime Compute for Apache Flink VVR 8.0.1 ou une version ultérieure prend en charge AnalyticDB for PostgreSQL V7.0.

Remarque
Si vous utilisez un connecteur personnalisé, consultez la section Gérer les connecteurs personnalisés pour plus d'informations.

Prérequis

Un espace de travail Fully Managed Flink est créé. Pour plus d'informations, consultez la section Activer Realtime Compute for Apache Flink.
Une instance AnalyticDB for PostgreSQL est créée. Pour plus d'informations, consultez la section Créer une instance.
L'instance AnalyticDB for PostgreSQL et l'espace de travail Fully Managed Flink doivent être déployés dans le même VPC.

Configurer une instance AnalyticDB for PostgreSQL

Connectez-vous à la console AnalyticDB for PostgreSQL.
Ajoutez le bloc CIDR de l'espace de travail Flink à la liste d'autorisation de l'instance AnalyticDB for PostgreSQL. Pour savoir comment configurer une liste d'autorisation, consultez la section Configurer une liste d'autorisation.
Cliquez sur Log On to Database. Pour connaître les autres méthodes de connexion à la base de données, consultez la section Se connecter à une instance à l'aide d'un client.
Créez une table sur l'instance AnalyticDB for PostgreSQL.
Exemple d'instruction SQL pour créer une table :
```
CREATE TABLE test_adbpg_table(
b1 int,
b2 int,
b3 text,
PRIMARY KEY(b1)
);
```

Configurer Realtime Compute for Flink

Connectez-vous à la console Realtime Compute.
Sous l'onglet Fully Managed Flink, cliquez sur Console dans la colonne Actions de l'espace de travail cible.
Dans le volet de navigation de gauche, cliquez sur Connectors.
Sur la page Connectors, cliquez sur Create Custom Connector.
Téléchargez le fichier JAR du connecteur personnalisé.
Remarque
- Obtenez le package JAR du connecteur Flink personnalisé pour AnalyticDB for PostgreSQL. Pour plus d'informations, consultez la page AnalyticDB PostgreSQL Connector.
- La version du package JAR doit correspondre à la version du moteur Flink de la plateforme Realtime Compute.
Une fois le téléchargement terminé, cliquez sur Next.

Le système analyse le connecteur personnalisé téléchargé. Si l'analyse réussit, passez à l'étape suivante. En cas d'échec, vérifiez que le code du connecteur personnalisé téléchargé est conforme aux normes de la communauté Apache Flink.
Cliquez sur Finish.

Le connecteur personnalisé créé apparaît dans la liste des connecteurs.

Créer une tâche Flink

Connectez-vous à la console Realtime Compute. Sous l'onglet Fully Managed Flink, cliquez sur Console dans la colonne Actions de l'espace de travail cible.
Dans le volet de navigation de gauche, cliquez sur SQL Development. Cliquez sur New, sélectionnez Blank Streaming Job Draft, puis cliquez sur Next.

Dans la boîte de dialogue New File Draft, configurez les paramètres de la tâche.

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

Cliquez sur Create.

Rédigez le code de la tâche.

Créez une table source aléatoire datagen_source et une table de destination test_adbpg_table dans AnalyticDB for PostgreSQL. Copiez le code de tâche suivant dans l'éditeur de texte de la tâche.

CREATE TABLE datagen_source (
 f_sequence INT,
 f_random INT,
 f_random_str STRING
) WITH (
 'connector' = 'datagen',
 'rows-per-second'='5',
 'fields.f_sequence.kind'='sequence',
 'fields.f_sequence.start'='1',
 'fields.f_sequence.end'='1000',
 'fields.f_random.min'='1',
 'fields.f_random.max'='1000',
 'fields.f_random_str.length'='10'
);
CREATE TABLE test_adbpg_table (
    `B1` bigint   ,
    `B2` bigint  ,
    `B3` VARCHAR ,
    `B4` VARCHAR,
     PRIMARY KEY(B1) not ENFORCED
) with (
   'connector' = 'adbpg-nightly-1.13',
   'password' = 'xxx',
   'tablename' = 'test_adbpg_table',
   'username' = 'xxxx',
   'url' = 'jdbc:postgresql://url:5432/schema',
   'maxretrytimes' = '2',
   'batchsize' = '50000',
   'connectionmaxactive' = '5',
   'conflictmode' = 'ignore',
   'usecopy' = '0',
   'targetschema' = 'public',
   'exceptionmode' = 'ignore',
   'casesensitive' = '0',
   'writemode' = '1',
   'retrywaittime' = '200'
);

Vous n'avez pas besoin de modifier les paramètres de la table datagen_source. Adaptez les paramètres de la table test_adbpg_table en fonction de vos besoins métier réels. Le tableau suivant décrit ces paramètres.

Paramètre	Obligatoire	Description
connector	Oui	Nom du connecteur. Définissez ce paramètre sur `adbpg-nightly-<numéro de version>`, par exemple `adbpg-nightly-1,13`.
url	Oui	URL JDBC de l'instance AnalyticDB for PostgreSQL. Format : `jdbc:postgresql://<point de terminaison interne>:<port>/<nom de la base de données>`. Exemple : `jdbc:postgresql://gp-xxxxxx.gpdb.cn-chengdu.rds.aliyuncs.com:5432/postgres`.
tablename	Oui	Nom de la table AnalyticDB for PostgreSQL.
username	Oui	Compte de base de données de l'instance AnalyticDB for PostgreSQL.
password	Oui	Mot de passe du compte de base de données de l'instance AnalyticDB for PostgreSQL.
maxretrytimes	Non	Nombre maximal de tentatives après un échec d'exécution SQL. Valeur par défaut : 3.
batchsize	Non	Nombre maximal d'enregistrements de données à écrire en un seul lot. Valeur par défaut : 50000.
exceptionmode	Non	Politique de gestion des erreurs en cas d'exception lors de l'écriture des données. Valeurs valides : ignore : ignore les données provoquant une exception. Il s'agit de la valeur par défaut. strict : déclenche un basculement et signale une erreur en cas d'exception d'écriture de données.
conflictmode	Non	Politique de gestion des conflits de clé primaire ou d'index unique. Valeurs valides : ignore : ignore les conflits de clé primaire et conserve les données existantes. strict : déclenche un basculement et signale une erreur en cas de conflit de clé primaire. update : met à jour les données en cas de conflit de clé primaire. upsert : écrit les données à l'aide de la méthode UPSERT en cas de conflit de clé primaire. Il s'agit de la valeur par défaut. AnalyticDB for PostgreSQL implémente UPSERT à l'aide des instructions INSERT ON CONFLICT et COPY ON CONFLICT. Si la table de destination est une table partitionnée, la version mineure du noyau doit être la V6.3.6.1 ou une version ultérieure. Pour savoir comment mettre à niveau la version mineure du noyau, consultez la section Mettre à niveau la version du moteur.
targetschema	Non	Schéma de la base de données AnalyticDB for PostgreSQL. Valeur par défaut : public.
writemode	Non	Mode d'écriture des données. Valeurs valides : 0 : écrit les données à l'aide de BATCH INSERT. 1 : écrit les données à l'aide de l'API COPY. Il s'agit de la valeur par défaut. 2 : écrit les données à l'aide de BATCH UPSERT.
verbose	Non	Indique si les journaux d'exécution du connecteur doivent être générés. Valeurs valides : 0 : ne génère pas de journaux d'exécution. Il s'agit de la valeur par défaut. 1 : génère des journaux d'exécution.
retrywaittime	Non	Intervalle entre les tentatives en cas d'exception. Unité : millisecondes. Valeur par défaut : 100.
batchwritetimeoutms	Non	Durée maximale d'accumulation du lot pour l'écriture par lots. Lorsque ce délai est dépassé, le lot accumulé est écrit. Unité : millisecondes. Valeur par défaut : 50000.
connectionmaxactive	Non	Paramètre du pool de connexions. Ce paramètre spécifie le nombre maximal de connexions simultanées dans le pool de connexions d'un seul Task Manager. Valeur par défaut : 5.
casesensitive	Non	Indique si les noms de colonnes et de tables sont sensibles à la casse. Valeurs valides : 0 : insensible à la casse. Il s'agit de la valeur par défaut. 1 : sensible à la casse.

Remarque

Les mappages de paramètres et de types sont pris en charge. Pour plus d'informations, consultez la documentation du connecteur pour AnalyticDB for PostgreSQL (ADB PG).

Démarrez la tâche.
1. En haut de la page de développement de la tâche, cliquez sur Deploy. Dans la boîte de dialogue qui s'affiche, cliquez sur OK.
  
  Remarque
  Les clusters de session conviennent au développement et aux tests dans des environnements hors production. Vous pouvez utiliser les clusters de session pour déboguer les tâches, améliorer l'utilisation des ressources du Job Manager (JM) et accélérer le démarrage des tâches. Toutefois, il n'est pas recommandé de soumettre des tâches aux clusters de session en raison de préoccupations liées à la stabilité métier. Pour plus d'informations, consultez la section Déboguer une tâche.
2. Sur la page Deployments, cliquez sur Resume dans la colonne Actions de la tâche cible.
3. Cliquez sur Resume.

Vérifier les résultats

Connectez-vous à la base de données AnalyticDB for PostgreSQL. Pour plus d'informations, consultez la section Se connecter à une instance à l'aide d'un client.
Exécutez l'instruction suivante pour interroger la table test_adbpg_table.
```
SELECT * FROM test_adbpg_table;
```
Les données sont écrites dans AnalyticDB for PostgreSQL comme prévu. La figure suivante présente un exemple de résultat.

Le résultat de la requête contient trois colonnes : b1 (int4), b2 (int4) et b3 (text). Plusieurs lignes de données sont renvoyées, ce qui confirme que les données ont été synchronisées avec la table de destination.

AnalyticDB:Écrire des données dans AnalyticDB for PostgreSQL