Use Paimon tables in EMR Serverless Spark - E-MapReduce - Centre de documentation Alibaba Cloud

Prérequis

Un espace de travail a été créé. Créer un espace de travail.

Procédure

Étape 1 : Créer une session SQL

Accédez à la page Sessions.
1. Connectez-vous à la console EMR.
2. Dans le volet de navigation de gauche, sélectionnez EMR Serverless > Spark.
3. Sur la page Spark, cliquez sur le nom de l'espace de travail cible.
4. Sur la page EMR Serverless Spark, cliquez sur Sessions dans le volet de navigation de gauche.
Sur la page SQL Session, cliquez sur Connect to SQL Session.

Sur la page Create SQL Session, configurez les paramètres dans la section Spark Configuration puis cliquez sur Create. Gérer les sessions SQL.

Spark lit et écrit des tables Paimon via des catalogues. Sélectionnez un type de catalogue adapté à votre scénario. Gérer les catalogues de données.

Catalogue de données

Lors de l'utilisation d'un catalogue de données, aucun paramètre de session n'est requis. Cliquez sur Catalog sur la page Add Catalog, puis sélectionnez le catalogue de données dans votre environnement de développement Spark SQL.

Remarque

Nous recommandons les versions de moteur EMR esr-4.3.0 ou ultérieure, esr-3.3.0 ou ultérieure, ou esr-2.7.0 ou ultérieure.

Catalogue personnalisé

Data Lake Formation (DLF)

spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       rest
spark.sql.catalog.<catalogName>.uri                             http://cn-hangzhou-vpc.dlf.aliyuncs.com
spark.sql.catalog.<catalogName>.warehouse                       <catalog_name>
spark.sql.catalog.<catalogName>.token.provider                  dlf
spark.sql.catalog.<catalogName>.dlf.access-key-id               <access_key_id>
spark.sql.catalog.<catalogName>.dlf.access-key-secret           <access_key_secret>

Paramètres :

Parameter	Description	Example value
`spark.sql.catalog.<catalogName>`	L'implémentation du catalogue.	Valeur fixe : `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	La méthode de stockage des métadonnées. Définissez-la sur `rest` pour utiliser l'API REST DLF.	Valeur fixe : `rest`
`spark.sql.catalog.<catalogName>.uri`	L'URI DLF. Format : `http://<endpoint>-vpc.dlf.aliyuncs.com`.	`http://cn-hangzhou-vpc.dlf.aliyuncs.com`
`spark.sql.catalog.<catalogName>.warehouse`	Le chemin du warehouse. Pour DLF, définissez cette valeur sur le nom du catalogue.	`<catalog_name>`
`spark.sql.catalog.<catalogName>.token.provider`	Le fournisseur d'authentification. Définissez-le sur `dlf` pour DLF.	Valeur fixe : `dlf`
`spark.sql.catalog.<catalogName>.dlf.access-key-id`	L'ID AccessKey de votre compte Alibaba Cloud ou utilisateur RAM.	`<access_key_id>`
`spark.sql.catalog.<catalogName>.dlf.access-key-secret`	Le secret AccessKey de votre compte Alibaba Cloud ou utilisateur RAM.	`<access_key_secret>`

DLF-Legacy

Les métadonnées sont stockées dans DLF 1.0.

spark.sql.catalog.<catalogName>                          org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                dlf
spark.sql.catalog.<catalogName>.dlf.catalog.id           <catalog_name>
spark.sql.catalog.<catalogName>.dlf.catalog.endpoint     dlf-vpc.cn-hangzhou.aliyuncs.com

Paramètres :

Parameter	Description	Example value
`spark.sql.catalog.<catalogName>`	L'implémentation du catalogue.	Valeur fixe : `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	La méthode de stockage des métadonnées. Définissez-la sur `dlf` pour utiliser DLF comme metastore.	Valeur fixe : `dlf`
`spark.sql.catalog.<catalogName>.dlf.catalog.id`	Le nom du catalogue dans DLF.	`<catalog_name>`
`spark.sql.catalog.<catalogName>.dlf.catalog.endpoint`	L'endpoint DLF. Sélectionnez un endpoint DLF en fonction de votre région.	`dlf-vpc.cn-hangzhou.aliyuncs.com`

Hive Metastore

Les métadonnées sont stockées dans un Hive Metastore spécifié.

spark.sql.catalog.<catalogName>                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore       hive
spark.sql.catalog.<catalogName>.uri             thrift://<yourHMSUri>:<port>

Paramètres :

Parameter	Description	Example value
`spark.sql.catalog.<catalogName>`	L'implémentation du catalogue.	Valeur fixe : `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	Le type de metastore. Définissez-le sur `hive` pour utiliser un Hive Metastore.	Valeur fixe : `hive`
`spark.sql.catalog.<catalogName>.uri`	L'URI Hive Metastore. Format : `thrift://<IP address of Hive metastore>:9083`. `<IP address of Hive metastore>` correspond à l'adresse IP interne du service Hive Metastore. Pour utiliser un metastore externe, consultez la rubrique Se connecter à un service Hive Metastore externe.	`thrift://192.168..:9083`

Système de fichiers

Les métadonnées sont stockées dans un système de fichiers.

spark.sql.catalog.<catalogName>                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore       filesystem
spark.sql.catalog.<catalogName>.warehouse       oss://<yourBucketName>/warehouse

Paramètres :

Parameter	Description	Example value
`spark.sql.catalog.<catalogName>`	L'implémentation du catalogue.	Valeur fixe : `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	Le type de metastore. Définissez-le sur `filesystem` pour utiliser un système de fichiers comme metastore.	Valeur fixe : `filesystem`
`spark.sql.catalog.<catalogName>.warehouse`	Le chemin du warehouse. Dans cet exemple, `<yourBucketName>` représente le nom d'un bucket OSS.	`oss://my-bucket/warehouse`

Vous pouvez configurer plusieurs catalogues simultanément, tels que DLF, DLF 1.0 et Hive. Exemple :

# Configure a DLF catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       rest
spark.sql.catalog.<catalogName>.uri                             http://cn-hangzhou-vpc.dlf.aliyuncs.com
spark.sql.catalog.<catalogName>.warehouse                       <catalog_name>
spark.sql.catalog.<catalogName>.token.provider                  dlf
spark.sql.catalog.<catalogName>.dlf.access-key-id               <access_key_id>
spark.sql.catalog.<catalogName>.dlf.access-key-secret           <access_key_secret>

# Configure a DLF 1.0 catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       dlf
spark.sql.catalog.<catalogName>.dlf.catalog.id                  <catalog_name>
spark.sql.catalog.<catalogName>.dlf.catalog.endpoint            dlf-vpc.cn-hangzhou.aliyuncs.com

# Configure a hive1 catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       hive
spark.sql.catalog.<catalogName>.uri                             thrift://<yourHMSUri-1>:<port>

# Configure a hive2 catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       hive
spark.sql.catalog.<catalogName>.uri                             thrift://<yourHMSUri-2>:<port>

Étape 2 : Lire et écrire des tables Paimon

Accédez à la page de développement SQL.

Sur la page EMR Serverless Spark, cliquez sur Development dans le volet de navigation de gauche.
Sur l'onglet Development, cliquez sur l'icône .
Dans la boîte de dialogue Create, saisissez un nom tel que users_task, définissez le Type sur la valeur par défaut SparkSQL, puis cliquez sur OK.

Copiez le code suivant dans le nouvel onglet Spark SQL (users_task).

Catalogue Paimon

-- Create a database.
CREATE DATABASE IF NOT EXISTS paimon.ss_paimon_db;             

-- Create a Paimon table.
CREATE TABLE paimon.ss_paimon_db.paimon_tbl (id INT, name STRING) USING paimon;

-- Write data to the Paimon table.
INSERT INTO paimon.ss_paimon_db.paimon_tbl VALUES (1, "a"), (2, "b"), (3, "c");

-- Query the write results from the Paimon table.
SELECT * FROM paimon.ss_paimon_db.paimon_tbl ORDER BY id;

-- Delete the database.
DROP DATABASE paimon.ss_paimon_db CASCADE;

Sélectionnez une base de données dans la liste déroulante des bases de données et la session SQL créée dans la liste déroulante des sessions.
Cliquez sur Run pour exécuter la tâche. La tâche renvoie la sortie suivante.

FAQ

Pourquoi une erreur s'affiche-t-elle lors de l'exécution d'une instruction DELETE, UPDATE ou MERGE sur une table ?

Symptôme : L'exécution d'une instruction DELETE, UPDATE ou MERGE renvoie une erreur similaire à la suivante :

Caused by: org.apache.spark.sql.AnalysisException: Table does not support deletes/updates/merge: <tableName>.
    at org.apache.spark.sql.errors.QueryCompilationErrors$.tableDoesNotSupportError(QueryCompilationErrors.scala:1391)

Cause : Le format de stockage de la table ne prend pas en charge les mises à jour au niveau des lignes, ou une configuration Spark requise est manquante.
Solution :
1. Vérifiez le type de table.
  
  Exécutez la commande suivante pour vérifier si la table est une table Paimon :
```
SHOW CREATE TABLE <tableName>;
```
  Si la sortie contient USING PAIMON, il s'agit d'une table Paimon. Si la sortie indique un autre format de stockage (tel que USING hive), vérifiez que le format prend en charge les mises à jour au niveau des lignes.
2. Vérifiez la configuration Spark.
  
  Si la table est une table Paimon, vérifiez la section Spark Configuration et assurez-vous que la configuration suivante est présente :
```
spark.sql.extensions org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
```
  Si elle est manquante, ajoutez-la à la section Spark Configuration.

Documentation connexe

Démarrage rapide pour le développement SparkSQL
Pour plus de détails sur l'utilisation et la configuration de Paimon, consultez la documentation officielle de Paimon.
Pour utiliser un service de metastore externe, consultez la rubrique Se connecter à un service Hive Metastore externe.