Connectez EMR Serverless Spark à un catalogue Data Lake Formation (DLF) via le protocole REST Iceberg pour lire et écrire des tables Iceberg avec Spark SQL.
Prérequis
Avant de commencer, vérifiez que vous disposez des éléments suivants :
Un espace de travail Serverless Spark créé dans la même région que votre instance DLF. Voir Créer un espace de travail
Types de tâches pris en charge
Les trois types de tâches permettent de se connecter à DLF via le catalogue REST Iceberg :
| Type de tâche | Référence |
|---|---|
| Session SQL | Gérer les sessions SQL |
| Spark Thrift Server | Gérer les sessions Spark Thrift Server |
| Tâche par lots | Développer une tâche par lots |
Étape 1 : Accorder des autorisations sur le catalogue
Connectez-vous à la console Data Lake Formation.
Sur la page Catalogs, cliquez sur le nom du catalogue pour afficher sa page de détails.
Cliquez sur l'onglet Permissions pour accorder l'accès à l'ensemble du catalogue. Pour accorder l'accès à une base de données ou à une table spécifique, accédez à cette ressource et cliquez sur son onglet Permissions.
-
Configurez les champs suivants et cliquez sur OK :
RemarqueSi AliyunECSInstanceForEMRRole n'apparaît pas dans la liste déroulante, accédez à la page de gestion des utilisateurs et cliquez sur Sync.
Champ Valeur User/Role Sélectionnez RAM User/RAM Role Select Authorization Object Sélectionnez AliyunECSInstanceForEMRRole dans la liste déroulante Preset Permission Type Sélectionnez manuellement les autorisations de lecture ou choisissez un rôle prédéfini tel que Data Reader ou Data Editor
Si vous êtes un utilisateur Resource Access Management (RAM), accordez les autorisations nécessaires sur les ressources avant d'effectuer des opérations sur les données. Voir Gestion des autorisations de données.
Étape 2 : Se connecter au catalogue et lire/écrire des données
Choisissez l'une des méthodes de connexion suivantes en fonction de la manière dont vous gérez votre catalogue.
Option 1 : Utiliser un catalogue de données (recommandé)
Si vous utilisez un catalogue de données géré par DLF, aucune configuration de session Spark n'est requise. Accédez à la page Data Catalog, cliquez sur Add data catalog, puis sélectionnez directement le catalogue lors du développement Spark SQL.
Option 2 : Utiliser un catalogue personnalisé
Ajoutez la configuration suivante dans la section Spark Configuration de Custom Configuration.
La configuration ci-dessous utilise iceberg_catalog comme nom de catalogue. Ce nom enregistre un service de gestion de tables Iceberg dans Spark, s'appuyant sur le catalogue REST Iceberg qui se connecte à DLF via des API REST. Modifiez le nom du catalogue et les paramètres associés selon vos besoins.
# Enable the Iceberg Spark extension
spark.sql.extensions org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
# Register the catalog
spark.sql.catalog.iceberg_catalog org.apache.iceberg.spark.SparkCatalog
# Use the Iceberg REST catalog implementation
spark.sql.catalog.iceberg_catalog.catalog-impl org.apache.iceberg.rest.RESTCatalog
# DLF Iceberg REST endpoint
spark.sql.catalog.iceberg_catalog.uri http://${regionID}-vpc.dlf.aliyuncs.com/iceberg
# Your DLF catalog name
spark.sql.catalog.iceberg_catalog.warehouse ${catalogName}
# Use the DLF FileIO implementation
spark.sql.catalog.iceberg_catalog.io-impl org.apache.iceberg.rest.DlfFileIO
# Enable SigV4 signature authentication
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
# Access credentials
spark.sql.catalog.iceberg_catalog.rest.access-key-id ${access_key_id}
spark.sql.catalog.iceberg_catalog.rest.secret-access-key ${access_key_secret}Remplacez les espaces réservés suivants par vos valeurs réelles :
| Espace réservé | Description | Exemple |
|---|---|---|
${regionID} |
ID de la région où votre instance DLF est déployée. Voir Endpoints. | cn-hangzhou |
${catalogName} |
Nom de votre catalogue DLF | my-catalog |
${access_key_id} |
ID de clé d'accès (AccessKey ID) de votre compte Alibaba Cloud | — |
${access_key_secret} |
Secret de clé d'accès (AccessKey secret) de votre compte Alibaba Cloud | — |
Pour les sessions SQL, utilisez la version du moteur esr-4.7.0, esr-3.6.0 ou ultérieure.
Lire et écrire des données
Les exemples suivants illustrent des opérations Spark SQL courantes sur une table Iceberg dans DLF. Toutes les instructions font référence aux tables au format iceberg_catalog.<database>.<table>.
Si aucune base de données n'est spécifiée, les tables sont créées dans la base de données default du catalogue.
Pour une présentation complète du développement Spark SQL, voir Prise en main du développement Spark SQL.
-- Create a database
CREATE DATABASE IF NOT EXISTS db;
-- Create a non-partitioned table
CREATE TABLE iceberg_catalog.db.tbl (
id BIGINT NOT NULL COMMENT 'unique id',
data STRING
)
USING iceberg;
-- Insert rows
INSERT INTO iceberg_catalog.db.tbl VALUES
(1, 'Alice'),
(2, 'Bob'),
(3, 'Charlie');
-- Query all rows
SELECT * FROM iceberg_catalog.db.tbl;
-- Query by condition
SELECT * FROM iceberg_catalog.db.tbl WHERE id = 2;
-- Update a row
UPDATE iceberg_catalog.db.tbl SET data = 'David' WHERE id = 3;
-- Confirm the update
SELECT * FROM iceberg_catalog.db.tbl WHERE id = 3;
-- Delete a row
DELETE FROM iceberg_catalog.db.tbl WHERE id = 1;
-- 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
)
USING iceberg
PARTITIONED BY (category);
-- Insert rows
INSERT INTO iceberg_catalog.db.part_tbl VALUES
(100, 'Data1', 'A', to_timestamp('2025-01-01 12:00:00')),
(200, 'Data2', 'B', to_timestamp('2025-01-02 14:00:00')),
(300, 'Data3', 'A', to_timestamp('2025-01-01 15:00:00')),
(400, 'Data4', 'C', to_timestamp('2025-01-03 10:00:00'));
-- Query all rows
SELECT * FROM iceberg_catalog.db.part_tbl;
-- Filter by bucket
SELECT * FROM iceberg_catalog.db.part_tbl WHERE bucket(16, id) = 0;
-- Filter by day
SELECT * FROM iceberg_catalog.db.part_tbl WHERE days(ts) = '2025-01-01';
-- Filter by partition column
SELECT * FROM iceberg_catalog.db.part_tbl WHERE category = 'A';
-- Combined filter (bucket + day + category)
SELECT * FROM iceberg_catalog.db.part_tbl
WHERE bucket(16, id) = 0
AND days(ts) = '2025-01-01'
AND category = 'A';
-- Aggregate by category
SELECT category, COUNT(*) AS count
FROM iceberg_catalog.db.part_tbl
GROUP BY category;
-- Drop the database (all tables must be empty first)
-- DROP DATABASE iceberg_catalog.db;