Realtime Compute for Apache Flink:Gérer les catalogues Iceberg

Prérequis

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

• Un espace de travail Flink exécutant Ververica Runtime (VVR) 11.1 ou une version ultérieure

• Un catalogue DLF dans la même région que votre espace de travail Flink. Consultez Démarrage rapide avec DLF

Limitations

Créer un catalogue Iceberg DLF

La création d'un catalogue Iceberg dans Flink enregistre un mappage vers votre catalogue DLF existant. Cette opération ne crée ni ne modifie les données dans DLF. Toutes les tables créées dans le catalogue DLF via Iceberg REST sont des tables Iceberg.

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

2. Dans la colonne Actions de votre espace de travail, cliquez sur Console.

3. Dans le volet de navigation de gauche, accédez à Development > Scripts et créez un nouveau script.

4. Dans l'éditeur SQL, collez l'instruction suivante. Dans le coin inférieur droit, cliquez sur Environment et sélectionnez un cluster de session exécutant VVR 11.2.0 ou une version ultérieure. Exécutez ensuite l'instruction.

   CREATE CATALOG `catalog_name`
   WITH (
       'type'              = 'iceberg',
       'catalog-type'      = 'rest',
       'token.provider'    = 'dlf',
       'uri'               = 'http://cn-hangzhou-vpc.dlf.aliyuncs.com/iceberg',
       'warehouse'         = 'iceberg_test',
       'rest.signing-region' = 'cn-hangzhou',
       'io-impl'           = 'org.apache.iceberg.rest.DlfFileIO'
   );

   Remplacez les valeurs de uri, warehouse et rest.signing-region par vos propres paramètres. Consultez Point de terminaison Iceberg REST pour obtenir la liste des URI disponibles par région. Le tableau suivant décrit chaque option.

   Option	Obligatoire	Description	Exemple
   type	Oui	Définissez la valeur sur iceberg.	iceberg
   catalog-type	Oui	Définissez la valeur sur rest.	rest
   token.provider	Oui	Définissez la valeur sur dlf. Configure l'authentification basée sur DLF.	dlf
   uri	Oui	Le point de terminaison Iceberg REST pour votre catalogue DLF. L'URI varie selon la région. Consultez Point de terminaison Iceberg REST.	http://ap-southeast-1-vpc.dlf.aliyuncs.com/iceberg
   warehouse	Oui	Le nom de votre catalogue DLF.	iceberg_test
   rest.signing-region	Oui	L'ID de la région où DLF est déployé.	cn-hangzhou
   io-impl	Oui	Définissez la valeur sur org.apache.iceberg.rest.DlfFileIO.	org.apache.iceberg.rest.DlfFileIO

Afficher ou supprimer un catalogue

1. Sur la console Realtime Compute for Apache Flink, cliquez sur Console dans la colonne Actions de votre espace de travail.

2. Accédez à la page Data Management et ouvrez Catalog List.

   • Pour afficher les bases de données et les tables d'un catalogue, cliquez sur View à côté du nom du catalogue.

   • Pour supprimer un catalogue, cliquez sur Delete dans la colonne Actions.

     Important

     La suppression d'un catalogue retire uniquement son enregistrement de Data Management dans votre projet Flink. Les fichiers de données Iceberg sous-jacents dans DLF ne sont pas affectés. Pour réutiliser les tables, enregistrez à nouveau le catalogue.

   Vous pouvez également exécuter l'instruction suivante sur la page Data Query :

   DROP CATALOG <catalog_name>;

Gérer les bases de données Iceberg

Exécutez les instructions suivantes sur la page Data Query : sélectionnez le code et cliquez sur Run.

-- Replace my-catalog with your catalog name.
USE CATALOG `my-catalog`;

-- Replace my_db with the database name you want to create.
CREATE DATABASE `my_db`;

-- Replace my-catalog with your catalog name.
USE CATALOG `my-catalog`;

-- Delete a database only if it is empty.
DROP DATABASE `my_db`;

-- Delete a database and all tables inside it.
DROP DATABASE `my_db` CASCADE;

Gérer les tables Iceberg

Une fois un catalogue Iceberg enregistré, référencez ses tables dans un job sans déclarer de langage de définition de données (DDL). Utilisez le nom complet en trois parties <catalog>.<database>.<table>, ou exécutez d'abord USE CATALOG et USE <database>, puis référencez la table par son nom seul.

-- Replace my-catalog with your Iceberg catalog name.
-- Replace my_db with your database name.
-- Replace my_tbl with your table name.
CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  dt           STRING,
  shop_id      BIGINT,
  user_id      BIGINT,
  num_orders   INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt);

ALTER TABLE my_table RENAME TO my_table_new;

-- Add c1 (INT) and c2 (STRING) to the end of the table.
ALTER TABLE my_table ADD (c1 INT, c2 STRING);

-- Rename column c0 to c1.
ALTER TABLE my_table RENAME c0 TO c1;

-- Delete columns c1 and c2.
ALTER TABLE my_table DROP (c1, c2);

ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count';

-- Move col_a to the beginning of the table.
ALTER TABLE my_table MODIFY col_a DOUBLE FIRST;

-- Move col_a immediately after col_b.
ALTER TABLE my_table MODIFY col_a DOUBLE AFTER col_b;

-- Temporarily enable upsert mode when writing to a table.
INSERT INTO tableName /*+ OPTIONS('upsert-enabled'='true') */
SELECT ...;

-- Temporarily set the source read interval to 10 seconds.
SELECT * FROM my_table /*+ OPTIONS('monitor-interval'='10s') */;

-- Replace my-catalog, my_db, and my_tbl with your values.
DROP TABLE `my-catalog`.`my_db`.`my_tbl`;