Ingest data to DLF with Flink DataStream APIs - Data Lake Formation

Prérequis

Vous disposez d'un espace de travail entièrement géré pour Realtime Compute for Apache Flink. Si vous n'en avez pas encore créé, consultez la rubrique Activer Realtime Compute for Apache Flink.
Un catalogue DLF a été créé. Pour plus d'informations, consultez la rubrique Premiers pas avec DLF.
Votre espace de travail Flink et votre catalogue DLF se trouvent dans la même région.
Le VPC de votre espace de travail Flink figure sur la liste d'autorisation VPC de DLF. Pour plus d'informations, consultez la rubrique Guide d'utilisation de l'API.

Remarque
DLF active l'accès VPC par défaut. Pour activer l'accès au réseau public, consultez la rubrique Activer et configurer l'accès au réseau public.

Préparatifs

Téléchargez le fichier JAR groupé Paimon paimon-flink-*.jar en version 1.1 ou ultérieure depuis le site web Apache Paimon.
Téléchargez le fichier paimon-oss-*.jar en version 1.1 ou ultérieure depuis la page Apache Paimon Filesystems.

Choisir une méthode de dépendance

Important

L'environnement d'exécution Flink n'inclut ni le connecteur Paimon ni le système de fichiers OSS. Vous devez utiliser l'une des méthodes suivantes pour vous assurer que les fichiers paimon-flink-*.jar et paimon-oss-*.jar sont disponibles lors de l'exécution de la tâche.

Méthode 1 : Télécharger des fichiers supplémentaires dans la console

Aucune modification du fichier pom.xml n'est nécessaire. Lorsque vous créez une tâche JAR dans la console de développement Realtime Compute for Apache Flink, téléchargez les fichiers paimon-flink-*.jar et paimon-oss-*.jar que vous avez obtenus lors des Préparatifs en tant que fichiers de dépendance supplémentaires.

Méthode 2 : Empaqueter les dépendances dans un fat JAR avec Maven

Ajoutez les dépendances et propriétés suivantes au fichier pom.xml de votre projet.

<properties>
    <!-- Paimon version. Specify 1.1 or later. -->
    <paimon.version>1.1.0</paimon.version>
    <!-- Flink major version. Set based on your VVR version. See the following table. -->
    <flink.main.version>1.20</flink.main.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.paimon</groupId>
        <artifactId>paimon-flink-${flink.main.version}</artifactId>
        <version>${paimon.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.paimon</groupId>
        <artifactId>paimon-oss</artifactId>
        <version>${paimon.version}</version>
    </dependency>
</dependencies>

La valeur de ${flink.main.version} est indiquée ci-dessous.

Version VVR

flink.main.version

VVR 8.x

1,17

VVR 11.x

1,20

Avec cette méthode, les dépendances sont intégrées au fat JAR ; il n'est donc pas nécessaire de télécharger des fichiers JAR supplémentaires lors du déploiement.

Étape 1 : Écrire le code de la tâche

Dans la méthode main() de votre tâche DataStream, utilisez le code suivant pour créer une instance de catalogue DLF.

Options options = new Options();
options.set("type", "paimon");
options.set("metastore", "rest");
options.set("uri", "http://<region-id>-vpc.dlf.aliyuncs.com");
options.set("warehouse", "your-catalog-name");
options.set("token.provider", "dlf");
options.set("dlf.access-key-id", "your-access-key-id");
options.set("dlf.access-key-secret", "your-access-key-secret");
Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(options);

Paramètres obligatoires :

Paramètre

Description

Exemple

type

Type de catalogue, analysé automatiquement à partir du fichier JAR personnalisé. Ne modifiez pas cette valeur.

paimon

metastore

Type de metastore pour DLF. Définissez ce paramètre sur rest.

rest

uri

Endpoint VPC du serveur de catalogue REST DLF. Le format est http://[region-id]-vpc.dlf.aliyuncs.com. Pour plus d'informations, consultez la rubrique Régions et endpoints.

http://ap-southeast-1-vpc.dlf.aliyuncs.com

warehouse

Nom du catalogue Paimon.

dlf_test

token.provider

Fournisseur de jeton. Définissez ce paramètre sur dlf.

dlf

dlf.access-key-id

Votre ID AccessKey pour l'authentification. Pour plus d'informations, consultez la rubrique Afficher les informations AccessKey d'un utilisateur RAM.

dlf.access-key-secret

Votre secret AccessKey pour l'authentification.

Une fois le catalogue créé, vous pouvez l'enregistrer et l'utiliser dans votre tâche DataStream pour lire et écrire des tables Paimon.

Étape 2 : Empaqueter et déployer la tâche

Empaquetez votre tâche DataStream dans un fichier JAR.
Téléchargez le fichier JAR de la tâche dans la console Realtime Compute for Apache Flink et soumettez la tâche.
Si vous avez choisi la méthode 1 (téléchargement de fichiers supplémentaires dans la console), ajoutez les fichiers paimon-flink-*.jar et paimon-oss-*.jar aux dépendances supplémentaires lors de la soumission de la tâche.

Pour plus d'informations sur le développement et le débogage des tâches JAR Flink, consultez la rubrique Développer une tâche JAR.