Use Spark to write data to an Iceberg table and read data from the table in batch mode - E-MapReduce

Utilice la API DataFrame de Spark para escribir datos en una tabla Iceberg y leerlos en modo por lotes dentro de un clúster de Hadoop de E-MapReduce (EMR). Este tema utiliza Spark 3.x.

Requisitos previos

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

Un clúster de Hadoop de EMR que ejecute EMR V3.38.0, EMR V5.4.0 o una versión secundaria posterior. Para obtener más información, consulte Create a cluster.

Paso 1: Agregar dependencias de Maven

Cree un proyecto de Maven y agregue las dependencias de Spark e Iceberg al archivo del modelo de objetos del proyecto (POM).

En el siguiente ejemplo se agregan las dependencias de Spark 3.1.1 e Iceberg 0.12.0 con el alcance provided. Esto significa que las dependencias se compilan de manera local, pero se resuelven desde el clúster durante la ejecución:

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.12</artifactId>
    <version>3.1.1</version>
    <scope>provided</scope>
</dependency>

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId>
    <version>3.1.1</version>
    <scope>provided</scope>
</dependency>

<dependency>
    <groupId>org.apache.iceberg</groupId>
    <artifactId>iceberg-core</artifactId>
    <version>0.12.0</version>
    <scope>provided</scope>
</dependency>

El paquete de Iceberg incluido en su clúster de EMR es distinto de la dependencia de Iceberg de código abierto. Por ejemplo, la integración con el catálogo de Data Lake Formation (DLF) ya viene incorporada. Utilice la dependencia de Iceberg de código abierto con el alcance provided para compilar y empaquetar el código de forma local. Después, permita que el paquete incluido en el clúster tome el control durante la ejecución.

Paso 2: Configurar un catálogo

Antes de invocar la API DataFrame de Spark sobre una tabla Iceberg, configure un catálogo en el objeto SparkConf. El nombre del catálogo y los parámetros obligatorios varían según la versión del clúster. Todos los ejemplos siguientes utilizan DLF para gestionar los metadatos. Para conocer los detalles de cada parámetro, consulte Configuration of DLF metadata.

Seleccione la configuración correspondiente a la versión de su clúster:

EMR V3.40 o posterior, EMR V5.6.0 o posterior: nombre del catálogo: iceberg

sparkConf.set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
sparkConf.set("spark.sql.catalog.iceberg", "org.apache.iceberg.spark.SparkCatalog")
sparkConf.set("spark.sql.catalog.iceberg.catalog-impl", "org.apache.iceberg.aliyun.dlf.hive.DlfCatalog")

EMR V3.39.X, EMR V5.5.X: nombre del catálogo: dlf

sparkConf.set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
sparkConf.set("spark.sql.catalog.dlf", "org.apache.iceberg.spark.SparkCatalog")
sparkConf.set("spark.sql.catalog.dlf.catalog-impl", "org.apache.iceberg.aliyun.dlf.hive.DlfCatalog")
sparkConf.set("spark.sql.catalog.dlf.warehouse", "<yourOSSWarehousePath>")

EMR V3.38.X, EMR V5.3.X, EMR V5.4.X: nombre del catálogo: dlf_catalog

Establezca las variables de entorno ALIBABA_CLOUD_ACCESS_KEY_ID y ALIBABA_CLOUD_ACCESS_KEY_SECRET antes de ejecutar el código. Para ver las instrucciones, consulte la sección Configuración de variables de entorno .

sparkConf.set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
sparkConf.set("spark.sql.catalog.dlf_catalog", "org.apache.iceberg.spark.SparkCatalog")
sparkConf.set("spark.sql.catalog.dlf_catalog.catalog-impl", "org.apache.iceberg.aliyun.dlf.DlfCatalog")
sparkConf.set("spark.sql.catalog.dlf_catalog.io-impl", "org.apache.iceberg.hadoop.HadoopFileIO")
sparkConf.set("spark.sql.catalog.dlf_catalog.oss.endpoint", "<yourOSSEndpoint>")
sparkConf.set("spark.sql.catalog.dlf_catalog.warehouse", "<yourOSSWarehousePath>")
sparkConf.set("spark.sql.catalog.dlf_catalog.access.key.id", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
sparkConf.set("spark.sql.catalog.dlf_catalog.access.key.secret", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"))
sparkConf.set("spark.sql.catalog.dlf_catalog.dlf.catalog-id", "<yourCatalogId>")
sparkConf.set("spark.sql.catalog.dlf_catalog.dlf.endpoint", "<yourDLFEndpoint>")
sparkConf.set("spark.sql.catalog.dlf_catalog.dlf.region-id", "<yourDLFRegionId>")

Paso 3: Escribir datos en una tabla Iceberg

Utilice la API DataFrameWriterV2 (y no la DataFrameWriterV1) para escribir datos de Spark 3.x en una tabla Iceberg. Se recomienda la API v2 porque cada operación se asigna directamente a una instrucción SQL y el comportamiento de sobrescritura es explícito:

Operación de DataFrame	SQL equivalente	Descripción
`df.writeTo(t).create()`	`CREATE TABLE AS SELECT`	Crea una tabla nueva a partir del DataFrame
`df.writeTo(t).replace()`	`REPLACE TABLE AS SELECT`	Reemplaza una tabla existente
`df.writeTo(t).createOrReplace()`	`CREATE OR REPLACE TABLE AS SELECT`	Crea la tabla si no existe o la reemplaza
`df.writeTo(t).append()`	`INSERT INTO`	Agrega filas a una tabla existente
`df.writeTo(t).overwritePartitions()`	`INSERT OVERWRITE` dinámico	Sobrescribe las particiones coincidentes

En los siguientes ejemplos, sustituya <yourCatalogName> por el nombre real de su catálogo.

Crear una tabla:

val df: DataFrame = ...
df.writeTo("<yourCatalogName>.iceberg_db.sample").create()

Al crear la tabla, utilice tableProperty() para establecer sus propiedades y partitionedBy() para definir los campos de partición.

Agregar datos:

val df: DataFrame = ...
df.writeTo("<yourCatalogName>.iceberg_db.sample").append()

Sobrescribir particiones:

val df: DataFrame = ...
df.writeTo("<yourCatalogName>.iceberg_db.sample").overwritePartitions()

Paso 4: Leer datos de una tabla Iceberg

Seleccione el método de lectura según su versión de Spark:

Spark 3.x (recomendado)

val df = spark.table("<yourCatalogName>.iceberg_db.sample")

Spark 2.4

val df = spark.read.format("iceberg").load("<yourCatalogName>.iceberg_db.sample")

Ejemplo: lectura y escritura por lotes de extremo a extremo

Este ejemplo utiliza un clúster EMR V5.3.0 con el catálogo dlf_catalog. El nombre del catálogo y sus parámetros cambian según la versión del clúster; para más detalles, consulte Configuration of DLF metadata.

Paso 1: Crear una base de datos de Iceberg

Utilice Spark SQL para crear una base de datos llamada iceberg_db. Para ver las instrucciones, consulte Use Iceberg.

Paso 2: Escribir la aplicación de Spark

El siguiente código en Scala configura el catálogo dlf_catalog, crea una tabla Iceberg, agrega datos y lee el resultado:

def main(args: Array[String]): Unit = {

  // Configure the DLF catalog
  val sparkConf = new SparkConf()
  sparkConf.set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
  sparkConf.set("spark.sql.catalog.dlf_catalog", "org.apache.iceberg.spark.SparkCatalog")
  sparkConf.set("spark.sql.catalog.dlf_catalog.catalog-impl", "org.apache.iceberg.aliyun.dlf.DlfCatalog")
  sparkConf.set("spark.sql.catalog.dlf_catalog.io-impl", "org.apache.iceberg.hadoop.HadoopFileIO")
  sparkConf.set("spark.sql.catalog.dlf_catalog.oss.endpoint", "<yourOSSEndpoint>")
  sparkConf.set("spark.sql.catalog.dlf_catalog.warehouse", "<yourOSSWarehousePath>")
  sparkConf.set("spark.sql.catalog.dlf_catalog.access.key.id", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
  sparkConf.set("spark.sql.catalog.dlf_catalog.access.key.secret", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"))
  sparkConf.set("spark.sql.catalog.dlf_catalog.dlf.catalog-id", "<yourCatalogId>")
  sparkConf.set("spark.sql.catalog.dlf_catalog.dlf.endpoint", "<yourDLFEndpoint>")
  sparkConf.set("spark.sql.catalog.dlf_catalog.dlf.region-id", "<yourDLFRegionId>")

  val spark = SparkSession
    .builder()
    .config(sparkConf)
    .appName("IcebergReadWriteTest")
    .getOrCreate()

  // Create the Iceberg table and write the first batch of rows
  val firstDF = spark.createDataFrame(Seq(
    (1, "a"), (2, "b"), (3, "c")
  )).toDF("id", "data")

  firstDF.writeTo("dlf_catalog.iceberg_db.sample").createOrReplace()

  // Append a second batch of rows
  val secondDF = spark.createDataFrame(Seq(
    (4, "d"), (5, "e"), (6, "f")
  )).toDF("id", "data")

  secondDF.writeTo("dlf_catalog.iceberg_db.sample").append()

  // Read all rows from the Iceberg table
  val icebergTable = spark.table("dlf_catalog.iceberg_db.sample")

  icebergTable.show()
}

Sustituya los siguientes marcadores de posición por los valores reales:

Marcador de posición	Descripción
`<yourOSSEndpoint>`	Endpoint de OSS; por ejemplo, `oss-cn-hangzhou.aliyuncs.com`
`<yourOSSWarehousePath>`	Ruta de OSS que se usa como raíz del almacén de Iceberg
`<yourCatalogId>`	ID del catálogo de DLF
`<yourDLFEndpoint>`	Endpoint de DLF
`<yourDLFRegionId>`	ID de la región donde se despliega DLF; por ejemplo, `cn-hangzhou`

Paso 3: Compilar y desplegar la aplicación

Agregue el plugin de Scala para Maven al archivo pom.xml para compilar los archivos fuente de Scala:

<build>
    <plugins>
        <plugin>
            <groupId>net.alchim31.maven</groupId>
            <artifactId>scala-maven-plugin</artifactId>
            <version>3.2.2</version>
            <executions>
                <execution>
                    <goals>
                        <goal>compile</goal>
                        <goal>testCompile</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

Depure el código de manera local y luego ejecute el siguiente comando para empaquetarlo:
```
mvn clean install
```
Inicie sesión en su clúster de EMR mediante SSH. Para ver las instrucciones, consulte Log on to a cluster.
Cargue el paquete JAR en el clúster de EMR. En este ejemplo, el paquete JAR se coloca en el directorio raíz del clúster.

Paso 4: Enviar el trabajo de Spark

Ejecute el siguiente comando spark-submit para enviar el trabajo:

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --driver-memory 1g \
  --executor-cores 1 \
  --executor-memory 1g \
  --num-executors 1 \
  --class com.aliyun.iceberg.IcebergTest \
  iceberg-demos.jar

Este ejemplo utiliza un paquete JAR llamado iceberg-demos.jar cuya clase principal es com.aliyun.iceberg.IcebergTest . Actualice --class y el nombre del archivo JAR para que coincidan con su proyecto.

Una vez que finaliza el trabajo, se devuelve el siguiente resultado:

+---+----+
| id|data|
+---+----+
|  4|   d|
|  1|   a|
|  5|   e|
|  6|   f|
|  2|   b|
|  3|   c|
+---+----+

Pasos siguientes

Use Iceberg: gestione tablas Iceberg con Spark SQL.
Configuration of DLF metadata: referencia de la configuración de catálogos para todas las versiones de EMR.