MaxCompute:Créer une table externe Apache Paimon à l'aide de Realtime Compute for Apache Flink

Informations générales

Apache Paimon est un format de stockage de lac de données unifié pour les flux continus et les traitements par lots, offrant des écritures à haut débit et des requêtes à faible latence. Les moteurs de calcul courants, tels que Spark, Hive et Trino dans Realtime Compute for Apache Flink et E-MapReduce, s'intègrent parfaitement à Paimon. Avec Apache Paimon, vous pouvez rapidement construire un lac de données sur Object Storage Service (OSS) et le connecter à MaxCompute pour l'analytique des lacs de données. Pour plus d'informations sur Apache Paimon, consultez la documentation Apache Paimon.

Prérequis

• Le compte Alibaba Cloud utilisé pour effectuer les opérations dispose de l'autorisation CreateTable permettant de créer des tables MaxCompute. Pour plus d'informations sur les autorisations de table, consultez la section Autorisations MaxCompute.

• Un projet MaxCompute a été créé. Pour plus d'informations, consultez la section Créer un projet.

• OSS est activé. Un bucket et un répertoire de fichiers ont été créés. Pour plus d'informations, consultez la section Créer un bucket.

  Remarque

  MaxCompute est déployé uniquement dans certaines régions. Pour éviter les problèmes de connectivité des données entre régions, nous vous recommandons d'utiliser un bucket situé dans la même région que votre projet MaxCompute.

• Fully managed Flink est activé. Pour plus d'informations, consultez la section Créer un espace de travail.

Étape 1 : Télécharger le plug-in Apache Paimon vers votre projet MaxCompute

Utilisez l'une des méthodes suivantes pour télécharger le plug-in Paimon vers votre projet MaxCompute.

ADD JAR <path_to_paimon_maxcompute_connector.jar>;

Étape 2 : Créer une table externe Apache Paimon à l'aide de Realtime Compute for Apache Flink

La procédure décrite dans cette rubrique s'appuie sur Realtime Compute for Apache Flink. Realtime Compute for Apache Flink écrit les données des fichiers Apache Paimon dans OSS. Un catalogue Apache Paimon est créé dans la console Realtime Compute for Apache Flink, et une table Apache Paimon permettant à MaxCompute de lire les données des fichiers Apache Paimon dans OSS est créée dans ce catalogue. Ensuite, MaxCompute utilise la table Apache Paimon comme table externe pour lire les données Apache Paimon stockées dans OSS.

1. Connectez-vous à la console Realtime Compute for Apache Flink et créez un Script. Pour plus d'informations, consultez la section Query Scripts.

2. Dans l'éditeur de script, saisissez la définition du catalogue et les valeurs des paramètres suivantes. Sélectionnez le code et cliquez sur Run.

   CREATE CATALOG `<catalog name>` WITH (
    'type' = 'paimon',
     'metastore' = 'maxcompute',
     'warehouse' = '<warehouse>',
     'maxcompute.endpoint' = '<maxcompute.endpoint>',
     'maxcompute.project' = '<maxcompute.project>',
     'maxcompute.accessid' = '<maxcompute.accessid>',
     'maxcompute.accesskey' = '<maxcompute.accesskey>',
     'maxcompute.oss.endpoint' = '<maxcompute.oss.endpoint>',
     '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 du code.

   Paramètre	Obligatoire	Description
   catalog name	Oui	Nom du catalogue Apache Paimon. Le nom ne peut contenir que des lettres. Dans cette rubrique, le nom du catalogue est catalogname.
   type	Oui	Type du catalogue. Définissez la valeur sur paimon.
   metastore	Oui	Type du stockage des métadonnées. Définissez la valeur sur maxcompute.
   warehouse	Oui	Répertoire de l'entrepôt de données dans OSS, au format oss://<bucket>/<object>. bucket : nom du bucket OSS que vous avez créé. object : chemin d'accès où vos données sont stockées. Vous pouvez afficher les noms de vos buckets et objets dans la console OSS.
   maxcompute.endpoint	Oui	Endpoint du service MaxCompute. Vous devez configurer ce paramètre en fonction de la région et du type de connexion réseau sélectionnés lors de la création du projet MaxCompute. Pour plus d'informations sur les endpoints correspondant aux différentes régions et types de réseau, consultez la section Endpoint.
   maxcompute.project	Oui	Nom du projet MaxCompute. Les projets MaxCompute pour lesquels la fonctionnalité de schéma est activée ne sont pas pris en charge.
   maxcompute.accessid	Oui	AccessKey ID du compte Alibaba Cloud ou de l'utilisateur RAM disposant des autorisations sur MaxCompute. Vous pouvez obtenir l'AccessKey ID sur la page AccessKey Management.
   maxcompute.accesskey	Oui	AccessKey secret correspondant à l'AccessKey ID.
   maxcompute.oss.endpoint	Non	Endpoint OSS auquel MaxCompute accède. Si vous ne configurez pas ce paramètre, la valeur du paramètre fs.oss.endpoint est utilisée par défaut. Important Le bucket OSS se trouve dans la même région que le projet MaxCompute. Nous vous recommandons de définir le paramètre maxcompute.oss.endpoint sur un endpoint interne. Pour plus d'informations sur les endpoints OSS des différents types de réseau dans chaque région, consultez la section Régions et endpoints.
   fs.oss.endpoint	Non	Endpoint d'OSS. Ce paramètre est requis si le bucket OSS spécifié par le paramètre warehouse ne se trouve pas dans la même région que l'espace de travail Realtime Compute for Apache Flink ou si un bucket OSS appartenant à un autre compte Alibaba Cloud est utilisé. Remarque Vous devez configurer l'endpoint en fonction de la région et de la méthode de connexion réseau sélectionnées lors de la création du bucket OSS. Pour plus d'informations sur les endpoints correspondant aux différentes régions et types de réseau, consultez la section Régions et endpoints.
   fs.oss.accessKeyId	Non	AccessKey ID du compte Alibaba Cloud ou de l'utilisateur RAM disposant des autorisations de lecture et d'écriture sur OSS. Ce paramètre est requis si le bucket OSS spécifié par le paramètre warehouse ne se trouve pas dans la même région que l'espace de travail Realtime Compute for Apache Flink ou si un bucket OSS appartenant à un autre compte Alibaba Cloud est utilisé. Vous pouvez obtenir l'AccessKey ID sur la page AccessKey Management.
   fs.oss.accessKeySecret	Non	AccessKey secret correspondant à l'AccessKey ID. Ce paramètre est requis si le bucket OSS spécifié par le paramètre warehouse ne se trouve pas dans la même région que l'espace de travail Realtime Compute for Apache Flink ou si un bucket OSS appartenant à un autre compte Alibaba Cloud est utilisé.

3. Créez une table Apache Paimon.

   1. Créez une table nommée test_tbl.

      Sous l'onglet Script, exécutez l'instruction suivante dans l'éditeur de script. Attendez qu'un message de confirmation s'affiche sous l'onglet Result. Cet exemple utilise une table nommée test_tbl.

      CREATE TABLE `catalogname`.`default`.test_tbl (
       dt STRING,
       id BIGINT,
       data STRING,
       PRIMARY KEY (dt, id) NOT ENFORCED
      ) PARTITIONED BY (dt);

   2. Écrivez des données dans la table test_tbl.

      Sur la page Drafts, créez un job SQL contenant les instructions suivantes. Déployez et exécutez ensuite le job. Pour plus d'informations sur la création et l'exécution de jobs SQL, consultez la section Aperçu du développement de jobs.

      -- The checkpoint interval is set to 10s to commit data faster.
      SET 'execution.checkpointing.interval' = '10s';
      INSERT INTO `catalogname`.`default`.test_tbl VALUES ('2023-04-21', 1, 'AAA'), ('2023-04-21', 2, 'BBB'), ('2023-04-22', 1, 'CCC'), ('2023-04-22', 2, 'DDD');

      Remarque

      • La table de résultat Apache Paimon valide les données à chaque fin de checkpoint.

      • En environnement de production, l'intervalle de checkpoint et l'intervalle minimal entre les checkpoints varient selon vos exigences métier en matière de latence. Ils sont généralement définis entre 1 et 10 minutes.

      • La version du moteur du brouillon SQL doit être vvr-8.0.5-flink-1.17 ou ultérieure.

Étape 3 : Lire les données avec MaxCompute

1. Exécutez les commandes suivantes sur le client local (odpscmd) ou via un autre outil capable d'exécuter des instructions SQL MaxCompute :

   SET odps.sql.common.table.planner.ext.hive.bridge = true;
   SET odps.sql.hive.compatible = true;

2. Exécutez la commande suivante pour interroger les données de la table externe Apache Paimon test_tbl :

   SELECT * FROM test_tbl WHERE dt = '2023-04-21';

   Le résultat suivant est renvoyé :

   +------------+------------+------------+
   | id | data | dt |
   +------------+------------+------------+
   | 1 | AAA | 2023-04-21 |
   | 2 | BBB | 2023-04-21 |
   +------------+------------+------------+