Use Iceberg in EMR - E-MapReduce - Centre de documentation Alibaba Cloud

Apache Iceberg sur EMR permet d'exécuter des opérations INSERT, SELECT, UPDATE et DELETE sur des données tabulaires stockées dans OSS, avec des métadonnées gérées par Data Lake Formation (DLF). Cette rubrique explique comment configurer un catalogue Iceberg basé sur DLF et exécuter des opérations Spark SQL de base sur un cluster Hadoop.

Prérequis

Avant de commencer, vérifiez que vous disposez des éléments suivants :

Un cluster Hadoop exécutant EMR V5.3.0 ou une version ultérieure. Consultez la page Créer un cluster.
Un accès SSH au cluster. Consultez la procédure Se connecter à un cluster.

Limitations

Les extensions Spark SQL d'Iceberg ne sont pas prises en charge avec Spark 2.4. Sur les clusters exécutant EMR V3.38.X ou une version ultérieure, utilisez l'API Spark DataFrame plutôt que Spark SQL pour interagir avec Iceberg. Cette rubrique décrit l'approche Spark SQL pour EMR V5.3.0 et les versions ultérieures.

Configurer le catalogue Iceberg

Avant d'exécuter Spark SQL sur Iceberg, configurez un catalogue pointant vers DLF en tant que magasin de métadonnées. Le nom du catalogue et les paramètres requis varient selon la version du cluster. Toutes les configurations utilisent la même extension Spark SQL :

org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

La configuration du catalogue est préfixée par spark.sql.catalog.<catalog_name>, où <catalog_name> correspond au nom attribué au catalogue.

Résumé des versions :

Version du cluster	Nom du catalogue par défaut	Classe `catalog-impl`
EMR V5.6.0 ou ultérieur	`iceberg`	`org.apache.iceberg.aliyun.dlf.hive.DlfCatalog`
EMR V5.5.X	`dlf`	`org.apache.iceberg.aliyun.dlf.hive.DlfCatalog`
EMR V5.3.X à V5.4.X	`dlf_catalog`	`org.apache.iceberg.aliyun.dlf.DlfCatalog`

EMR V5.6.0 ou ultérieur

Connectez-vous à votre cluster via SSH, puis exécutez la commande suivante :

spark-sql --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
 --conf spark.sql.catalog.iceberg=org.apache.iceberg.spark.SparkCatalog \
 --conf spark.sql.catalog.iceberg.catalog-impl=org.apache.iceberg.aliyun.dlf.hive.DlfCatalog \

EMR V5.5.X

Connectez-vous à votre cluster via SSH, puis exécutez la commande suivante :

spark-sql --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
 --conf spark.sql.catalog.dlf=org.apache.iceberg.spark.SparkCatalog \
 --conf spark.sql.catalog.dlf.catalog-impl=org.apache.iceberg.aliyun.dlf.hive.DlfCatalog \
 --conf spark.sql.catalog.dlf.warehouse=<your-oss-warehouse-path> \

Vous pouvez laisser le paramètre spark.sql.catalog.dlf.warehouse vide. Si ce paramètre est omis, le chemin d'entrepôt par défaut est utilisé.

EMR V5.3.X à V5.4.X

Cette version nécessite une paire de clés AccessKey. Définissez les variables d'environnement ALIBABA_CLOUD_ACCESS_KEY_ID et ALIBABA_CLOUD_ACCESS_KEY_SECRET avant d'exécuter la commande. Pour plus d'instructions, consultez la page Configurer les variables d'environnement sous Linux, macOS et Windows.

Connectez-vous à votre cluster via SSH, puis exécutez la commande suivante :

spark-sql --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
 --conf spark.sql.catalog.dlf_catalog=org.apache.iceberg.spark.SparkCatalog \
 --conf spark.sql.catalog.dlf_catalog.catalog-impl=org.apache.iceberg.aliyun.dlf.DlfCatalog \
 --conf spark.sql.catalog.dlf_catalog.io-impl=org.apache.iceberg.hadoop.HadoopFileIO \
 --conf spark.sql.catalog.dlf_catalog.oss.endpoint=<your-oss-endpoint> \
 --conf spark.sql.catalog.dlf_catalog.warehouse=<your-oss-warehouse-path> \
 --conf spark.sql.catalog.dlf_catalog.access.key.id=<ALIBABA_CLOUD_ACCESS_KEY_ID> \
 --conf spark.sql.catalog.dlf_catalog.access.key.secret=<ALIBABA_CLOUD_ACCESS_KEY_SECRET> \
 --conf spark.sql.catalog.dlf_catalog.dlf.catalog-id=<your-catalog-id> \
 --conf spark.sql.catalog.dlf_catalog.dlf.endpoint=<your-dlf-endpoint> \
 --conf spark.sql.catalog.dlf_catalog.dlf.region-id=<your-dlf-region-id>

Remplacez les espaces réservés par vos valeurs réelles :

Espace réservé	Description
`<your-oss-endpoint>`	Point de terminaison OSS de votre région, par exemple `oss-cn-hangzhou.aliyuncs.com`
`<your-oss-warehouse-path>`	Chemin OSS où Iceberg stocke les données de table, par exemple `oss://my-bucket/warehouse`
`<ALIBABA_CLOUD_ACCESS_KEY_ID>`	Valeur de la variable d'environnement `ALIBABA_CLOUD_ACCESS_KEY_ID`
`<ALIBABA_CLOUD_ACCESS_KEY_SECRET>`	Valeur de la variable d'environnement `ALIBABA_CLOUD_ACCESS_KEY_SECRET`
`<your-catalog-id>`	ID du catalogue DLF
`<your-dlf-endpoint>`	Point de terminaison du service DLF
`<your-dlf-region-id>`	ID de la région où DLF est déployé, par exemple `cn-hangzhou`

Vérifier la session

Lorsque l'interface CLI Spark SQL est prête, l'invite de commande devient :

spark-sql>

Exécuter des opérations Iceberg de base

Les exemples ci-dessous utilisent <catalog_name> pour désigner le nom du catalogue configuré. Pour EMR V5.6.0 et les versions ultérieures, remplacez <catalog_name> par iceberg. Pour les autres versions, utilisez le nom du catalogue indiqué dans la section Configurer le catalogue Iceberg.

Créer une base de données

CREATE DATABASE IF NOT EXISTS <catalog_name>.iceberg_db;

Créer une table

La forme de base spécifie les noms et types de colonnes, ainsi que le moteur de stockage Iceberg :

CREATE TABLE IF NOT EXISTS <catalog_name>.iceberg_db.sample(
    id BIGINT COMMENT 'unique id',
    data STRING
)
USING iceberg;

La commande CREATE TABLE prend également en charge les clauses COMMENT, PARTITIONED BY, LOCATION et TBLPROPERTIES. Utilisez TBLPROPERTIES pour définir des propriétés au niveau de la table, telles que le format d'écriture par défaut :

CREATE TABLE IF NOT EXISTS <catalog_name>.iceberg_db.sample(
    id BIGINT COMMENT 'unique id',
    data STRING
)
USING iceberg
TBLPROPERTIES (
    'write.format.default'='parquet'
);

Insérer des données

INSERT INTO <catalog_name>.iceberg_db.sample VALUES (1, 'a'), (2, 'b'), (3, 'c');

Interroger des données

SELECT * FROM <catalog_name>.iceberg_db.sample;
SELECT count(1) AS count, data FROM <catalog_name>.iceberg_db.sample GROUP BY data;

Mettre à jour des données

UPDATE <catalog_name>.iceberg_db.sample SET data = 'x' WHERE id = 3;

Supprimer des données

DELETE FROM <catalog_name>.iceberg_db.sample WHERE id = 3;

Rubriques connexes

Pour plus de détails sur la configuration des métadonnées, consultez la page Configuration des métadonnées DLF.