E-MapReduce:Utiliser Iceberg

Prérequis

Un espace de travail est créé. Créer un espace de travail.

Étape 1 : Créer une ressource de session

1. Accédez à la page Sessions.

   1. Connectez-vous à la console EMR.

   2. Dans le volet de navigation de gauche, choisissez 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.

2. Sur la page SQL Session, cliquez sur Connect to SQL Session.

3. Sur la page Connect to SQL Session, configurez les paramètres suivants dans la section Spark Configuration, puis cliquez sur Create. Pour plus d'informations, consultez Gérer les sessions SQL.

   Spark lit et écrit dans les tables Iceberg via un catalogue. Choisissez un catalogue en fonction de votre cas d'utilisation. Pour plus d'informations sur les catalogues, consultez Gérer les catalogues de données.

   Catalogue de données

   Si vous utilisez un catalogue de données, aucun paramètre n'est requis dans la session. Sur la page Catalog, cliquez sur Add Catalog. Vous pourrez ensuite sélectionner directement le catalogue de données dans votre environnement de développement SparkSQL.

   Remarque

   • Pour accéder à Iceberg dans DLF (anciennement DLF 2.5), utilisez la version du moteur esr-4.7.0, esr-3.6.0 ou ultérieure.

   • Pour accéder à Iceberg dans DLF 1.0 (anciennement DLF-Legacy) ou Hive Metastore, utilisez la version du moteur esr-4.3.0, esr-3.3.0, esr-2.7.0 ou ultérieure.

   Catalogue personnalisé

   DLF (anciennement DLF 2.5)

   Remarque

   Version du moteur requise : esr-4.7.0, esr-3.6.0 ou ultérieure.

   spark.sql.extensions org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
   spark.sql.catalog.iceberg_catalog org.apache.iceberg.spark.SparkCatalog
   spark.sql.catalog.iceberg_catalog.catalog-impl org.apache.iceberg.rest.RESTCatalog
   spark.sql.catalog.iceberg_catalog.uri http://<regionID>-vpc.dlf.aliyuncs.com
   spark.sql.catalog.iceberg_catalog.warehouse  <catalog_name>
   spark.sql.catalog.iceberg_catalog.io-impl org.apache.iceberg.rest.DlfFileIO
   spark.sql.catalog.iceberg_catalog.rest.auth.type sigv4
   spark.sql.catalog.iceberg_catalog.rest.auth.sigv4.delegate-auth-type none
   spark.sql.catalog.iceberg_catalog.rest.signing-region <regionID>
   spark.sql.catalog.iceberg_catalog.rest.signing-name DlfNext
   spark.sql.catalog.iceberg_catalog.rest.access-key-id <access_key_id>
   spark.sql.catalog.iceberg_catalog.rest.secret-access-key <access_key_secret>

   Le tableau suivant décrit les paramètres.

   Paramètre	Description	Valeur d'exemple
   spark.sql.extensions	Active les extensions Iceberg pour Spark.	Valeur fixe : org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
   spark.sql.catalog.iceberg_catalog	Enregistre un catalogue Spark nommé iceberg_catalog.	Valeur fixe : org.apache.iceberg.spark.SparkCatalog
   spark.sql.catalog.iceberg_catalog.catalog-impl	Spécifie l'implémentation sous-jacente du catalogue comme étant le catalogue REST Iceberg.	org.apache.iceberg.rest.RESTCatalog
   spark.sql.catalog.iceberg_catalog.uri	Le point de terminaison API REST pour le service DLF Iceberg. Le format est http://<regionID>-vpc.dlf.aliyuncs.com.	http://cn-hangzhou-vpc.dlf.aliyuncs.com
   spark.sql.catalog.iceberg_catalog.warehouse	Le nom du catalogue DLF associé. Remarque N'associez pas un catalogue DLF créé à partir d'un partage de données.	<catalog_name>
   spark.sql.catalog.iceberg_catalog.io-impl	Utilise l'implémentation FileIO personnalisée par DLF.	Valeur fixe : org.apache.iceberg.rest.DlfFileIO
   spark.sql.catalog.iceberg_catalog.rest.auth.type	Active le mécanisme de signature AWS Signature Version 4 (SigV4) pour authentifier les requêtes REST.	sigv4
   spark.sql.catalog.iceberg_catalog.rest.auth.sigv4.delegate-auth-type	Désactive l'authentification déléguée. Le client doit fournir un AccessKey ID et un AccessKey Secret pour signer les requêtes.	none
   spark.sql.catalog.iceberg_catalog.rest.signing-region	La région utilisée pour la signature. Cette valeur doit correspondre à la région où le service DLF est déployé.	cn-hangzhou
   spark.sql.catalog.iceberg_catalog.rest.signing-name	Le nom du service utilisé pour la signature.	Valeur fixe : DlfNext
   spark.sql.catalog.iceberg_catalog.rest.access-key-id	L'AccessKey ID de votre compte Alibaba Cloud ou utilisateur RAM.	<access_key_id>
   spark.sql.catalog.iceberg_catalog.rest.secret-access-key	L'AccessKey Secret de votre compte Alibaba Cloud ou utilisateur RAM.	<access_key_secret>

   DLF 1.0 (anciennement DLF-Legacy)

   Remarque

   Version du moteur requise : esr-4.3.0, esr-3.3.0, esr-2.7.0 ou ultérieure.

   Les métadonnées sont stockées dans DLF 1.0 (anciennement DLF-Legacy).

   spark.sql.extensions                          org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
   spark.sql.catalog.<catalogName>               org.apache.iceberg.spark.SparkCatalog
   spark.sql.catalog.<catalogName>.catalog-impl  org.apache.iceberg.aliyun.dlf.hive.DlfCatalog
   spark.sql.catalog.<catalogName>.dlf.catalog.id <catalog_name>

   Le tableau suivant décrit les paramètres.

   Paramètre	Description	Valeur d'exemple
   spark.sql.extensions	Active les extensions Iceberg pour Spark.	Valeur fixe : org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
   spark.sql.catalog.<catalogName>	Enregistre un catalogue nommé <catalogName>.	Valeur fixe : org.apache.iceberg.spark.SparkCatalog
   spark.sql.catalog.<catalogName>.catalog-impl	Spécifie l'implémentation compatible Hive d'Alibaba Cloud DLF 1.0, qui se connecte au service de métadonnées DLF 1.0.	Valeur fixe : org.apache.iceberg.aliyun.dlf.hive.DlfCatalog
   spark.sql.catalog.<catalogName>.dlf.catalog.id	Spécifie le nom du catalogue DLF associé.	<catalog_name>

   Hive Metastore

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

   spark.sql.extensions                          org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
   spark.sql.catalog.<catalogName>               org.apache.iceberg.spark.SparkCatalog
   spark.sql.catalog.<catalogName>.catalog-impl  org.apache.iceberg.hive.HiveCatalog
   spark.sql.catalog.<catalogName>.uri           thrift://<yourHMSUri>:<port>

   Le tableau suivant décrit les paramètres.

   Paramètre	Description	Valeur d'exemple
   spark.sql.extensions	Active les extensions Iceberg pour Spark.	Valeur fixe : org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
   spark.sql.catalog.<catalogName>	Enregistre un catalogue nommé <catalogName>.	Valeur fixe : org.apache.iceberg.spark.SparkCatalog
   spark.sql.catalog.<catalogName>.catalog-impl	Indique que le catalogue utilise l'implémentation officielle Iceberg HiveCatalog pour gérer les métadonnées dans le Hive Metastore.	Valeur fixe : org.apache.iceberg.hive.HiveCatalog
   spark.sql.catalog.<catalogName>.uri	L'URI du Hive Metastore. Le format est thrift://<adresse IP du Hive metastore>:9083. <adresse IP du Hive metastore> correspond à l'adresse IP interne du service HMS. Si vous devez spécifier un service de metastore externe, consultez 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.extensions                          org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
   spark.sql.catalog.<catalogName>               org.apache.iceberg.spark.SparkCatalog
   spark.sql.catalog.<catalogName>.type          hadoop
   spark.sql.catalog.<catalogName>.warehouse     oss://<yourBucketName>/warehouse

   Le tableau suivant décrit les paramètres.

   Paramètre	Description	Valeur d'exemple
   spark.sql.extensions	Active les extensions Iceberg pour Spark.	Valeur fixe : org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
   spark.sql.catalog.<catalogName>	Enregistre un catalogue nommé <catalogName>.	Valeur fixe : org.apache.iceberg.spark.SparkCatalog
   spark.sql.catalog.<catalogName>.type	Spécifie le type de catalogue comme hadoop. HadoopCatalog stocke les métadonnées directement dans le système de fichiers sans nécessiter de Hive Metastore.	hadoop
   spark.sql.catalog.<catalogName>.warehouse	Spécifie le chemin de stockage des métadonnées. Dans ce chemin, <yourBucketName> représente le nom du bucket OSS.	oss://<yourBucketName>/warehouse

Étape 2 : Lire et écrire dans les tables Iceberg****

1. Accédez à la page Data Development.

   Sur la page EMR Serverless Spark, cliquez sur Development dans le volet de navigation de gauche.

2. Sur l'onglet Development, cliquez sur l'icône .

3. 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.

4. Copiez le code suivant dans le nouvel onglet SparkSQL (users_task).

   Remarque

   Si aucune base de données n'est spécifiée, la table est créée dans la base de données default du catalogue. Vous pouvez également créer et spécifier une autre base de données.

   -- Create a database.
   CREATE DATABASE IF NOT EXISTS iceberg_catalog.db;
   -- Create a non-partitioned table.
   CREATE TABLE iceberg_catalog.db.tbl (
       id BIGINT NOT NULL COMMENT 'unique id',
       data STRING
   )
   USING iceberg;
   -- Insert data into the non-partitioned table.
   INSERT INTO iceberg_catalog.db.tbl VALUES
   (1, 'Alice'),
   (2, 'Bob'),
   (3, 'Charlie');
   -- Query all data from the non-partitioned table.
   SELECT * FROM iceberg_catalog.db.tbl;
   -- Query the non-partitioned table with a condition.
   SELECT * FROM iceberg_catalog.db.tbl WHERE id = 2;
   -- Update data in the non-partitioned table.
   UPDATE iceberg_catalog.db.tbl SET data = 'David' WHERE id = 3;
   -- Query again to confirm the update.
   SELECT * FROM iceberg_catalog.db.tbl WHERE id = 3;
   -- Delete data from the non-partitioned table.
   DELETE FROM iceberg_catalog.db.tbl WHERE id = 1;
   -- Query again to confirm the deletion.
   SELECT * FROM iceberg_catalog.db.tbl;
   -- Create a partitioned table.
   CREATE TABLE iceberg_catalog.db.part_tbl (
       id BIGINT,
       data STRING,
       category STRING,
       ts TIMESTAMP,
       dt DATE
   )
   USING iceberg
   PARTITIONED BY (dt, category);
   -- Insert data into the partitioned table.
   INSERT INTO iceberg_catalog.db.part_tbl VALUES
     (1 , 'data-01', 'A', timestamp'2026-01-01 10:00:00', date'2026-01-01'),
     (2 , 'data-02', 'A', timestamp'2026-01-01 11:00:00', date'2026-01-01'),
     (3 , 'data-03', 'A', timestamp'2026-01-02 09:30:00', date'2026-01-02'),
     (4 , 'data-04', 'B', timestamp'2026-01-02 12:15:00', date'2026-01-02'),
     (5 , 'data-05', 'B', timestamp'2026-01-03 08:05:00', date'2026-01-03'),
     (6 , 'data-06', 'B', timestamp'2026-01-03 14:20:00', date'2026-01-03'),
     (7 , 'data-07', 'C', timestamp'2026-01-04 16:45:00', date'2026-01-04'),
     (8 , 'data-08', 'C', timestamp'2026-01-04 18:10:00', date'2026-01-04'),
     (9 , 'data-09', 'C', timestamp'2026-01-05 07:55:00', date'2026-01-05'),
     (10, 'data-10', 'A', timestamp'2026-01-05 13:35:00', date'2026-01-05');
   -- Query all data from the partitioned table.
   SELECT * FROM iceberg_catalog.db.part_tbl;
   -- Query data from the partition where dt='2026-01-01'.
   SELECT * FROM iceberg_catalog.db.part_tbl WHERE dt='2026-01-01';
   -- Query data for a specific category.
   SELECT * FROM iceberg_catalog.db.part_tbl WHERE category = 'A';
   -- Perform a query with multiple conditions (day + category).
   SELECT * FROM iceberg_catalog.db.part_tbl 
   WHERE dt='2026-01-01'
     AND category = 'A';
   -- Aggregate and count the number of records in each category.
   SELECT category, COUNT(*) AS count 
   FROM iceberg_catalog.db.part_tbl 
   GROUP BY category;
   -- Drop the database (use with caution). Before you drop the database, make sure that it contains no tables.
   -- DROP DATABASE iceberg_catalog.db;

5. Dans la liste déroulante des sessions, sélectionnez l'instance de session SQL que vous avez créée, puis cliquez sur Run. Une fois la requête d'agrégation terminée, la sortie affiche : catégorie B (3), catégorie C (3) et catégorie A (4).