Manage MaxCompute catalogs - Realtime Compute for Apache Flink

Fonctionnement

Le tableau suivant présente la correspondance entre les concepts MaxCompute et ceux des catalogues Flink :

Concept du catalogue Flink	Concept MaxCompute	Notes
Catalogue	—	Défini uniquement dans Flink ; représente la connexion à MaxCompute
Base de données	Projet MaxCompute	Une base de données par projet. Lorsque `catalog.schema.enabled` est défini sur `true`, une base de données correspond à un schéma MaxCompute
Table	Table physique MaxCompute	Les mappages de types de données entre les deux systèmes sont créés automatiquement. Aucune inscription DDL n'est nécessaire pour les tables existantes

Limites

Contrainte	Détails
Version du runtime	Requiert Ververica Runtime (VVR) 6.0.7 ou version ultérieure
Création de bases de données	Non prise en charge. Les bases de données d'un catalogue MaxCompute font référence à des projets MaxCompute, qui doivent déjà exister
Modification des schémas de table	Non prise en charge
CREATE TABLE AS (CTAS)	Non pris en charge. Consultez CTAS
Paramètres du catalogue	Impossible de les modifier après la création. Pour mettre à jour un paramètre, supprimez le catalogue et recréez-le

Prérequis

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

Un espace de travail Realtime Compute for Apache Flink
Un projet MaxCompute et son endpoint
Un compte Alibaba Cloud disposant des autorisations d'administrateur sur les projets MaxCompute auxquels le catalogue accédera
L'AccessKey ID et l'AccessKey secret de ce compte

Créer un catalogue MaxCompute

Deux méthodes sont disponibles : l'interface utilisateur de la console et Flink SQL. Privilégiez l'interface de la console, sauf si vous devez automatiser la création du catalogue via du code.

Interface utilisateur de la console (recommandée)

Connectez-vous à la console Realtime Compute for Apache Flink. Localisez votre espace de travail et cliquez sur Console dans la colonne Actions.
Dans le volet de navigation de gauche, cliquez sur Catalogs.
Sur la page Catalog List, cliquez sur Create Catalog. Dans la boîte de dialogue, sélectionnez ODPS et cliquez sur Next.

Configurez les paramètres du catalogue.

Paramètre	Description	Obligatoire
`catalog name`	Nom personnalisé du catalogue	Oui
`endpoint`	Endpoint MaxCompute. Consultez Endpoints	Oui
`accessId`	AccessKey ID du compte utilisé pour accéder à MaxCompute. Le compte doit disposer des autorisations d'administrateur sur les projets auxquels le catalogue accède	Oui
`accessKey`	AccessKey secret du compte	Oui
`project`	Projet MaxCompute à utiliser comme base de données par défaut. Si ce champ est laissé vide, le projet par défaut du compte est utilisé. Après la création, tous les projets de votre compte apparaissent dans les métadonnées du catalogue	Non
`catalog.schema.enabled`	Indique si la prise en charge des schémas est activée. `false` (par défaut) : une base de données Flink correspond à un projet MaxCompute. `true` : une base de données Flink correspond à un schéma MaxCompute. Activez cette option pour les projets utilisant la fonctionnalité de schéma	Non

Cliquez sur Confirm. Le nouveau catalogue apparaît dans le volet Catalogs sur le côté gauche de la page.

Si la paire de clés AccessKey ne dispose pas des autorisations nécessaires pour certaines tables ou bases de données dans MaxCompute, une erreur s'affiche. Cela n'empêche pas la lecture ou l'écriture des tables accessibles.

Flink SQL

Dans l'éditeur SQL, exécutez l'instruction suivante :

CREATE CATALOG `<catalogName>` WITH (
  'type' = 'odps',
  'endpoint' = '<odpsEndpoint>',
  'accessId' = '<aliyunAccountAccessId>',
  'accessKey' = '<aliyunAccountAccessKey>',
  'project' = '<defaultProject>',
  'userAccount' = '<RAMUserAccount>'
);

Paramètre	Description	Obligatoire
`catalogName`	Nom personnalisé du catalogue	Oui
`type`	Définissez la valeur sur `odps`	Oui
`endpoint`	Endpoint MaxCompute. Consultez Endpoints	Oui
`accessId`	AccessKey ID. Le compte doit disposer des autorisations d'administrateur sur les projets auxquels le catalogue accède	Oui
`accessKey`	AccessKey secret	Oui
`project`	Projet MaxCompute à utiliser comme base de données par défaut	Non
`userAccount`	Nom du compte Alibaba Cloud ou nom de l'utilisateur RAM. Définissez ce paramètre lorsque l'AccessKey secret appartient à un utilisateur RAM disposant des autorisations d'administrateur uniquement sur des projets spécifiques. Format : `RAM$[<account_name>:]<RAM_name>`. Le catalogue répertorie alors uniquement les projets auxquels le compte peut accéder. Consultez Planification et gestion des utilisateurs	Non

Sélectionnez l'instruction et cliquez sur Run sur le côté gauche de l'éditeur.

创建MaxCompute Catalog..png

Afficher un catalogue MaxCompute

Interface utilisateur de la console

Connectez-vous à la console Realtime Compute for Apache Flink. Localisez votre espace de travail et cliquez sur Console dans la colonne Actions.
Dans le volet de navigation de gauche, cliquez sur Catalogs.
Sur la page Catalog List, localisez le catalogue et vérifiez les colonnes Catalog Name et Type. Pour parcourir les bases de données et les tables, cliquez sur View dans la colonne Actions.

Flink SQL

Dans l'éditeur SQL, exécutez :

DESCRIBE `<catalogName>`.`<projectName>`.`<tableName>`;

Paramètre	Description
`catalogName`	Nom du catalogue
`projectName`	Nom du projet MaxCompute
`tableName`	Nom de la table physique MaxCompute

Sélectionnez l'instruction, cliquez sur Run, puis consultez le schéma de la table dans l'onglet Results situé sous l'éditeur.

Utiliser un catalogue MaxCompute

Créer une table physique MaxCompute

L'exécution d'une instruction DDL Flink CREATE TABLE sur un catalogue MaxCompute crée automatiquement la table physique correspondante dans MaxCompute et mappe les types de données entre les deux systèmes.

Important

Les noms de colonnes MaxCompute sont tous en minuscules. Les noms de colonnes dans les instructions DDL Flink sont sensibles à la casse ; les lettres majuscules sont automatiquement converties en minuscules. Si deux noms de colonnes deviennent identiques après cette conversion, l'instruction échoue.

Table non partitionnée

CREATE TABLE `<catalogName>`.`<projectName>`.`<tableName>` (
  f0 INT,
  f1 BIGINT,
  f2 DOUBLE,
  f3 STRING
);

Table partitionnée

Placez la colonne de clé de partition à la fin du schéma et déclarez-la dans PARTITIONED BY :

CREATE TABLE `<catalogName>`.`<projectName>`.`<tableName>` (
  f0 INT,
  f1 BIGINT,
  f2 DOUBLE,
  f3 STRING,
  ds STRING
) PARTITIONED BY (ds);

Dans cet exemple, f0, f1, f2 et f3 sont des colonnes régulières, tandis que ds est la clé de partition.

Lire les données d'une table de catalogue MaxCompute

Étant donné que le catalogue récupère automatiquement les schémas depuis MaxCompute, aucune déclaration DDL n'est requise avant la lecture.

Lire toutes les données (par défaut)

SELECT * FROM `<catalogName>`.`<projectName>`.`<tableName>`;

Par défaut, cette opération lit toutes les partitions d'une table partitionnée.

Lire une partition spécifique

SELECT * FROM `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds=230613') */;

Lire de manière incrémentielle (table source incrémentielle)

SELECT * FROM `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('startPartition' = 'ds=230613') */;

Utiliser comme table de dimension

SELECT * FROM `<anotherTable>` AS l LEFT JOIN
`<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'max_pt()', 'cache' = 'ALL') */
FOR SYSTEM_TIME AS OF l.proc_time AS r
ON l.id = r.id;

Pour connaître toutes les options disponibles, consultez Connecteur MaxCompute.

Ajouter un watermark

Les catalogues MaxCompute ne contiennent pas d'informations de watermark. Pour spécifier le temps d'événement lors de l'utilisation d'une table de catalogue comme source, utilisez CREATE TABLE ... LIKE ... :

CREATE TABLE `<newTable>` ( WATERMARK FOR ts AS ts )
LIKE `<catalogName>`.`<projectName>`.`<tableName>`;

Ici, ts est une colonne de type DATETIME dans la table physique MaxCompute. Après l'exécution de cette instruction, les lectures depuis <newTable> incluent les informations de watermark.

Écrire des données dans une table de catalogue MaxCompute

Les catalogues MaxCompute prennent en charge l'écriture dans des partitions statiques et dynamiques. Pour la liste complète des options de table de résultat, consultez Connecteur MaxCompute.

Important

Dans la liste SELECT, les colonnes de clé de partition doivent apparaître après les colonnes régulières, dans l'ordre des niveaux de partition.

Écrire dans une partition statique

-- Example: two-level partition (ds, hh), writing to a fixed partition
INSERT INTO `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds=20231024,hh=09') */
SELECT <otherColumns>, '20231024', '09' FROM `<anotherTable>`;

Écrire dans une partition dynamique

INSERT INTO `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds,hh') */
SELECT <otherColumns>, ds, hh FROM `<anotherTable>`;

Supprimer un catalogue MaxCompute

Avertissement

La suppression d'un catalogue n'affecte pas immédiatement les déploiements en cours d'exécution. Toutefois, tout déploiement faisant référence à une table du catalogue supprimé échouera lors de sa publication ou de son redémarrage. Procédez avec prudence.

Interface utilisateur de la console

Connectez-vous à la console Realtime Compute for Apache Flink. Localisez votre espace de travail et cliquez sur Console dans la colonne Actions.
Dans le volet de navigation de gauche, cliquez sur Catalogs.
Sur la page Catalog List, localisez le catalogue et cliquez sur Delete dans la colonne Actions.
Dans la boîte de dialogue de confirmation, cliquez sur Delete. Vérifiez que le catalogue n'apparaît plus dans le volet Catalogs.

Flink SQL

Dans l'éditeur SQL, exécutez :

DROP CATALOG `<catalogName>`;

Avertissement

Les déploiements en cours d'exécution ne sont pas affectés immédiatement, mais les brouillons non publiés et les déploiements nécessitant un cycle de suspension-reprise seront impactés.

Cliquez avec le bouton droit sur l'instruction et choisissez Run. Vérifiez que le catalogue n'apparaît plus dans le volet Catalogs.

Mappages de types de données

Pour la référence complète des types de données MaxCompute, consultez Système de types de données MaxCompute version 2.0.

De MaxCompute vers Flink

MaxCompute	Flink
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INTEGER
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(precision, scale)	DECIMAL(precision, scale)
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)
STRING	STRING
BINARY	BYTES
DATE	DATE
DATETIME	TIMESTAMP(3)
TIMESTAMP	TIMESTAMP(9)
ARRAY	ARRAY
MAP	MAP
STRUCT	ROW
JSON	STRING

De Flink vers MaxCompute

Flink	MaxCompute
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INTEGER	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(precision, scale)	DECIMAL(precision, scale)
CHAR(n)	CHAR(n)
VARCHAR / STRING	STRING
BINARY	BINARY
VARBINARY / BYTES	BINARY
DATE	DATE
TIMESTAMP(n<=3)	DATETIME
TIMESTAMP(3	TIMESTAMP
ARRAY	ARRAY
MAP	MAP
ROW	STRUCT