Realtime Compute for Apache Flink:Démarrage rapide Paimon : fonctionnalités de base

Prérequis

• Si vous utilisez un utilisateur RAM ou un rôle RAM, assurez-vous qu'il dispose des permissions nécessaires pour accéder à la console Realtime Compute for Apache Flink. Pour plus d'informations, consultez Gestion des permissions.

• Un espace de travail Flink doit être créé. Pour plus d'informations, consultez Activer Realtime Compute for Apache Flink.

• Object Storage Service (OSS) doit être activé et un bucket de classe de stockage Standard doit exister. Pour plus d'informations, consultez Premiers pas avec la console OSS. OSS sert à stocker les fichiers des tables Paimon, y compris les fichiers de données et de métadonnées.

• Les tables Paimon sont prises en charge uniquement à partir de Ververica Runtime (VVR) 8.0.5.

Étape 1 : Créer un catalogue Paimon

1. Accédez à l'éditeur de scripts.

   1. Connectez-vous à la console Realtime Compute for Apache Flink.

   2. Dans la colonne Actions de l'espace de travail cible, cliquez sur Console.

   3. Dans la barre de navigation de gauche, cliquez sur Data Development > Data Query. Dans l'onglet Query Script, créez un nouveau Query Script.

2. Saisissez le code suivant dans l'éditeur pour créer un catalogue Paimon.

   -- my-catalog is a custom catalog name.
   CREATE Catalog `my-catalog` WITH (
     'type' = 'paimon',
     'metastore' = 'filesystem',
     'warehouse' = '<warehouse>',
     'fs.oss.endpoint' = '<fs.oss.endpoint>',
     'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
     'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
   );

   Le tableau suivant décrit les paramètres.

   Paramètre	Description	Obligatoire	Remarques
   type	Type du catalogue.	Oui	La valeur doit être 'paimon'.
   metastore	Type de metastore.	Oui	Cet exemple utilise filesystem. Pour plus d'informations sur les autres types, consultez Gérer les catalogues Paimon.
   warehouse	Répertoire warehouse dans votre bucket OSS.	Oui	Le format est oss://<bucket>/<object>, où : bucket : nom du bucket OSS que vous avez créé. object : chemin où vous souhaitez stocker les données. Vous trouverez les noms de votre bucket et de votre objet dans la console OSS.
   fs.oss.endpoint	Endpoint du service OSS.	Non	Obligatoire si le bucket OSS spécifié par warehouse se trouve dans une région différente de celle de votre espace de travail Flink ou appartient à un autre compte Alibaba Cloud. Pour plus d'informations, consultez Régions et Endpoints. Remarque Pour stocker une table Paimon dans OSS-HDFS, vous devez spécifier les paramètres fs.oss.endpoint, fs.oss.accessKeyId et fs.oss.accessKeySecret. La valeur de fs.oss.endpoint suit le format cn-<region>.oss-dls.aliyuncs.com, par exemple cn-hangzhou.oss-dls.aliyuncs.com.
   fs.oss.accessKeyId	AccessKey ID d'un compte Alibaba Cloud ou d'un utilisateur RAM disposant des permissions de lecture et d'écriture sur OSS.	Non	Obligatoire si le bucket OSS spécifié par warehouse se trouve dans une région différente de celle de votre espace de travail Flink ou appartient à un autre compte Alibaba Cloud. Pour savoir comment obtenir une paire d'AccessKey, consultez Créer une paire d'AccessKey.
   fs.oss.accessKeySecret	AccessKey Secret du compte Alibaba Cloud ou de l'utilisateur RAM disposant des permissions de lecture et d'écriture sur OSS.	Non

3. Sélectionnez le code de création du catalogue Paimon et cliquez sur Run à gauche.

   Le message The following statement has been executed successfully! confirme la création du catalogue.

Étape 2 : Créer une base de données et une table Paimon

1. Sur la page Scripts, saisissez le code suivant pour créer une base de données Paimon nommée my_db et une table Paimon nommée my_tbl.

   CREATE DATABASE `my-catalog`.`my_db`;
   CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
     dt STRING,
     id BIGINT,
     content STRING,
     PRIMARY KEY (dt, id) NOT ENFORCED
   ) PARTITIONED BY (dt) WITH (
     'changelog-producer' = 'lookup'  
   );

   Remarque

   Pour permettre ultérieurement la consommation en streaming de la table Paimon, cet exemple définit changelog-producer sur lookup dans la clause WITH, ce qui utilise la stratégie lookup pour générer le journal des modifications. Pour plus d'informations, consultez Production du journal des modifications.

2. Sélectionnez le code créant la base de données et la table Paimon, puis cliquez sur Run à gauche.

   L'apparition du message The following statement has been executed successfully! indique que la base de données my_db et la table my_tbl ont bien été créées.

Étape 3 : Écrire des données dans la table Paimon

1. Dans l'onglet Drafts de la page Development > ETL, créez un brouillon de flux vide. Pour plus d'informations, consultez Développer un brouillon SQL. Copiez l'instruction INSERT suivante dans l'éditeur SQL.

   -- Paimon commits data only after each checkpoint is complete.
   -- The checkpoint interval is shortened to 10s to commit data faster.
   -- In a production environment, typically set the checkpoint interval and the minimum interval between checkpoints from 1 to 10 minutes, based on your latency requirements.
   SET 'execution.checkpointing.interval'='10s';
   INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108',1,'apple'), ('20240108',2,'banana'), ('20240109',1,'cat'), ('20240109',2,'dog');

2. Dans le coin supérieur droit de l'éditeur SQL, cliquez sur Deploy. Dans la boîte de dialogue Deploy draft, configurez les paramètres selon les instructions et cliquez sur Confirm.

3. Sur la page O&M > Deployments, localisez le déploiement cible et cliquez sur Start dans la colonne Actions. Sélectionnez Initial Mode puis cliquez sur Start.

   Lorsque l'état du job passe à FINISHED, les données ont été écrites.

Étape 4 : Consommer des données en streaming depuis la table Paimon

1. Créez un brouillon de flux vide. Copiez le code SQL suivant dans l'éditeur SQL pour envoyer toutes les données de la table my_tbl vers les journaux via le connecteur print.

   CREATE TEMPORARY TABLE Print (
     dt STRING,
     id BIGINT,
     content STRING
   ) WITH (
     'connector' = 'print'
   );
   INSERT INTO Print SELECT * FROM `my-catalog`.`my_db`.`my_tbl`;

2. Dans le coin supérieur droit de l'éditeur SQL, cliquez sur Deploy. Dans la boîte de dialogue Deploy draft, configurez les paramètres selon les instructions et cliquez sur Confirm.

3. Sur la page O&M > Deployments, cliquez sur Start dans la colonne Actions du déploiement cible. Sélectionnez Initial Mode puis cliquez sur Start.

4. Consultez les résultats de calcul Flink sur la page de détails du déploiement.

   1. Sur la page O&M > Deployments, cliquez sur le nom du déploiement cible.

   2. Dans l'onglet Logs, sous Running Logs, allez à l'onglet Running Task Managers et cliquez sur un lien de tâche dans la colonne Path, ID.

   3. Cliquez sur Stdout pour afficher les données Paimon consommées.

   La sortie Stdout affiche les enregistrements INSERT de la table Paimon : +I[20240108, 1, apple], +I[20240108, 2, banana], +I[20240109, 1, cat] et +I[20240109, 2, dog], ce qui confirme la réussite de la consommation en streaming.

Étape 5 : Mettre à jour les données de la table Paimon

1. Créez un brouillon de flux vide et copiez le code SQL suivant dans l'éditeur SQL.

   SET 'execution.checkpointing.interval' = '10s';
   INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108', 1, 'hello'), ('20240109', 2, 'world');

2. Dans le coin supérieur droit de l'éditeur SQL, cliquez sur Deploy. Dans la boîte de dialogue Deploy draft, configurez les paramètres selon les instructions et cliquez sur Confirm.

3. Sur la page O&M > Deployments, cliquez sur Start dans la colonne Actions du déploiement cible. Sélectionnez Initial Mode puis cliquez sur Start.

   Lorsque l'état du job passe à FINISHED, les données ont été mises à jour.

4. Dans l'onglet Stdout de la page Deployments correspondant au job de l'étape 4, visualisez les données mises à jour dans la table Paimon.

   Dans l'onglet Running Task Managers, cliquez sur le TaskManager cible et ouvrez le sous-onglet Stdout. Les enregistrements CDC de la table Paimon s'affichent : -U[20240108, 1, apple], +U[20240108, 1, hello], -U[20240109, 2, dog] et +U[20240109, 2, world]. -U correspond à l'enregistrement avant la mise à jour, tandis que +U correspond à l'enregistrement après la mise à jour.

Étape 6 (Facultatif) : Annuler le job et nettoyer les ressources

Après vos tests, annulez le job de consommation en streaming et nettoyez les ressources.

1. Sur la page O&M > Deployments, cliquez sur Cancel dans la colonne Actions du déploiement cible afin d'arrêter le job.

2. Dans l'éditeur de l'onglet Scripts, saisissez le code suivant pour supprimer les fichiers de données Paimon ainsi que le catalogue Paimon.

   DROP DATABASE `my-catalog`.`my_db` CASCADE; -- Deletes all data files for the Paimon database from OSS.
   DROP CATALOG `my-catalog`; -- Removes the Paimon catalog from the Realtime Compute for Apache Flink console metadata but does not affect the data files in OSS.

   Le message The following statement has been executed successfully! confirme la suppression des fichiers de données Paimon et du catalogue.

Rubriques connexes

• Pour plus d'informations sur l'écriture ou la consommation de données dans une table Paimon, consultez Écrire et consommer des données dans des tables Paimon.

• Pour modifier le schéma d'une table Paimon (ajout de colonnes, changement de types, etc.) ou ajuster temporairement ses paramètres, reportez-vous à Modifier un schéma de table.

• Concernant les optimisations courantes pour les tables à clé primaire Paimon et les tables Append Scalable, consultez Optimisation des performances Paimon.

• Pour les questions fréquentes relatives à Paimon, consultez Connecteurs.