Use Iceberg in EMR - E-MapReduce - Centro de documentación de Alibaba Cloud

Apache Iceberg en EMR permite ejecutar operaciones INSERT, SELECT, UPDATE y DELETE sobre datos con formato de tabla almacenados en OSS, con metadatos gestionados por Data Lake Formation (DLF). Este tema explica cómo configurar un catálogo de Iceberg respaldado por DLF y ejecutar operaciones básicas de Spark SQL en un clúster de Hadoop.

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Un clúster de Hadoop que ejecute EMR V5.3.0 o posterior. Consulte Crear un clúster
Acceso SSH al clúster. Consulte Log on to a cluster

Limitaciones

Las extensiones de Spark SQL para Iceberg no son compatibles con Spark 2.4. En clústeres que ejecutan EMR V3.38.X o posterior, utilice la API de DataFrame de Spark en lugar de Spark SQL para interactuar con Iceberg. Este tema aborda el enfoque de Spark SQL para EMR V5.3.0 y versiones posteriores.

Configuración del catálogo de Iceberg

Antes de ejecutar Spark SQL en Iceberg, configure un catálogo que apunte a DLF como almacén de metadatos. El nombre del catálogo y los parámetros obligatorios varían según la versión del clúster. Todas las configuraciones utilizan la misma extensión de Spark SQL:

org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

La configuración del catálogo lleva el prefijo spark.sql.catalog.<catalog_name>, donde <catalog_name> es el nombre asignado al catálogo.

Resumen de versiones:

Versión del clúster	Nombre predeterminado del catálogo	Clase `catalog-impl`
EMR V5.6.0 o posterior	`iceberg`	`org.apache.iceberg.aliyun.dlf.hive.DlfCatalog`
EMR V5.5.X	`dlf`	`org.apache.iceberg.aliyun.dlf.hive.DlfCatalog`
EMR V5.3.X a V5.4.X	`dlf_catalog`	`org.apache.iceberg.aliyun.dlf.DlfCatalog`

EMR V5.6.0 o posterior

Inicie sesión en su clúster mediante SSH y ejecute el siguiente comando:

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

Inicie sesión en su clúster mediante SSH y ejecute el siguiente comando:

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> \

Puede dejar spark.sql.catalog.dlf.warehouse en blanco. Si se omite, se utiliza la ruta de almacenamiento predeterminada.

EMR V5.3.X a V5.4.X

Esta versión requiere un par de AccessKey. Establezca las variables de entorno ALIBABA_CLOUD_ACCESS_KEY_ID y ALIBABA_CLOUD_ACCESS_KEY_SECRET antes de ejecutar el comando. Para obtener instrucciones, consulte Configurar variables de entorno en Linux, macOS y Windows.

Inicie sesión en su clúster mediante SSH y ejecute el siguiente comando:

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>

Reemplace los marcadores de posición con sus valores reales:

Marcador de posición	Descripción
`<your-oss-endpoint>`	Endpoint de OSS para su región; por ejemplo, `oss-cn-hangzhou.aliyuncs.com`
`<your-oss-warehouse-path>`	Ruta de OSS donde Iceberg almacena los datos de la tabla; por ejemplo, `oss://my-bucket/warehouse`
`<ALIBABA_CLOUD_ACCESS_KEY_ID>`	Valor de la variable de entorno `ALIBABA_CLOUD_ACCESS_KEY_ID`
`<ALIBABA_CLOUD_ACCESS_KEY_SECRET>`	Valor de la variable de entorno `ALIBABA_CLOUD_ACCESS_KEY_SECRET`
`<your-catalog-id>`	ID del catálogo de DLF
`<your-dlf-endpoint>`	Endpoint del servicio DLF
`<your-dlf-region-id>`	ID de la región donde se despliega DLF; por ejemplo, `cn-hangzhou`

Comprobación de la sesión

Cuando la CLI de Spark SQL esté lista, el símbolo del sistema cambiará a:

spark-sql>

Ejecución de operaciones básicas de Iceberg

Los siguientes ejemplos utilizan <catalog_name> para referirse al nombre del catálogo que configuró. Para EMR V5.6.0 y versiones posteriores, reemplace <catalog_name> por iceberg. Para otras versiones, utilice el nombre del catálogo indicado en la sección Configuración del catálogo de Iceberg.

Creación de una base de datos

CREATE DATABASE IF NOT EXISTS <catalog_name>.iceberg_db;

Creación de una tabla

La forma básica especifica los nombres de columna, los tipos y el motor de almacenamiento de Iceberg:

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

CREATE TABLE también admite las cláusulas COMMENT, PARTITIONED BY, LOCATION y TBLPROPERTIES. Utilice TBLPROPERTIES para establecer propiedades a nivel de tabla, como el formato de escritura predeterminado:

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

Inserción de datos

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

Consulta de datos

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

Actualización de datos

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

Eliminación de datos

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

Temas relacionados

Para más detalles sobre la configuración de metadatos, consulte Configuration of DLF metadata.