Write data to an Iceberg table by using Spark Structured Streaming. - E-MapReduce

Utilice Apache Spark Structured Streaming para escribir datos de forma continua desde Apache Kafka hacia una tabla Apache Iceberg en EMR. Todas las operaciones de escritura se realizan a través de la API DataStreamWriter.

Requisitos previos

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

Un clúster DataLake o un clúster personalizado. Para obtener más información, consulte Create a cluster.
Un clúster Dataflow con el servicio Kafka habilitado. Para obtener más información, consulte Create a cluster.
Ambos clústeres desplegados en el mismo vSwitch de la misma virtual private cloud (VPC).

Funcionamiento

Envíe un trabajo de escritura en streaming mediante la llamada a DataStreamWriter desde Spark Structured Streaming:

val tableIdentifier: String = ...
data.writeStream
    .format("iceberg")
    .outputMode("append")
    .trigger(Trigger.ProcessingTime(1, TimeUnit.MINUTES))
    .option("path", tableIdentifier)
    .option("checkpointLocation", checkpointPath)
    .start()

tableIdentifier corresponde al nombre o a la ruta de la tabla de metadatos de destino. Se admiten dos modos de salida:

Modo	Comportamiento	Equivalente SQL
`append`	Escribe cada micro-lote como filas nuevas	`INSERT INTO`
`complete`	Reemplaza la tabla completa con el micro-lote más reciente	`INSERT OVERWRITE`

Transmisión de datos desde Kafka a una tabla Iceberg

Este ejemplo lee datos de un clúster Dataflow (Kafka) y los escribe en una tabla Iceberg. Después de empaquetar el código, envíelo como un trabajo de Spark mediante spark-submit.

Paso 1: Configuración de un tema de Kafka con datos de prueba

Inicie sesión en el clúster Dataflow mediante SSH. Para obtener más información, consulte Log on to a cluster.

Cree un tema llamado iceberg_test:

kafka-topics.sh --bootstrap-server core-1-1:9092,core-1-2:9092,core-1-3:9092 \
  --topic iceberg_test \
  --partitions 3 \
  --replication-factor 2 \
  --create

Genere datos de prueba para el tema:

kafka-console-producer.sh --broker-list core-1-1:9092,core-1-2:9092,core-1-3:9092 \
  --topic iceberg_test

Paso 2: Creación de una base de datos y una tabla en Iceberg

Utilice Spark SQL para crear una base de datos llamada iceberg_db y una tabla llamada iceberg_table. Para obtener más información, consulte Use Iceberg.

Paso 3: Configuración del proyecto Maven

Cree un proyecto Maven, agregue las dependencias de Spark e incluya el plugin Maven de Scala para la compilación. Agregue lo siguiente a su archivo pom.xml:

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>3.1.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>3.1.2</version>
    </dependency>
</dependencies>
<build>
    <plugins>
        <!-- the Maven Scala plugin will compile Scala source files -->
        <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>

Paso 4: Desarrollo de la aplicación Spark

El siguiente código en Scala lee datos desde Kafka y los escribe en una tabla Iceberg en modo streaming.

Importante

Los parámetros del catálogo y los nombres predeterminados varían según la versión del clúster. Este ejemplo utiliza un clúster EMR V5.3.0 con DLF (Data Lake Formation) para gestionar los metadatos bajo un catálogo llamado dlf_catalog. Para conocer los detalles de configuración de DLF, consulte Configuration of DLF metadata.

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

  // Configure the Iceberg catalog backed by DLF.
  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", "<yourAccessKeyId>")
  sparkConf.set("spark.sql.catalog.dlf_catalog.access.key.secret", "<yourAccessKeySecret>")
  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("StructuredSinkIceberg")
    .getOrCreate()

  val checkpointPath = "oss://mybucket/tmp/iceberg_table_checkpoint"
  val bootstrapServers = "192.168.XX.XX:9092"
  val topic = "iceberg_test"

  // Read from the Kafka topic on the Dataflow cluster.
  val df = spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", bootstrapServers)
    .option("subscribe", topic)
    .load()

  val resDF = df.selectExpr(
      "CAST(unbase64(CAST(key AS STRING)) AS STRING) AS strKey", // Decode Base64-encoded key to string.
      "CAST(value AS STRING) AS data")
    .select(
      col("strKey").cast(LongType).alias("id"), // Cast string key to LONG type.
      col("data")
    )

  // Write to the Iceberg table every minute.
  val query = resDF.writeStream
    .format("iceberg")
    .outputMode("append")
    .trigger(Trigger.ProcessingTime(1, TimeUnit.MINUTES))
    .option("path", "dlf_catalog.iceberg_db.iceberg_table")
    .option("checkpointLocation", checkpointPath)
    .start()

  query.awaitTermination()
}

Puede modificar los valores de los parámetros descritos en la siguiente tabla según sus necesidades empresariales.

Variable	Descripción
`checkpointPath`	Ruta de OSS para los datos de checkpoint de Spark Structured Streaming
`bootstrapServers`	Dirección IP privada de un broker de Kafka en el clúster Dataflow
`topic`	Nombre del tema de Kafka al que se suscribirá

Paso 5: Empaquetado y despliegue en EMR

Después de realizar pruebas locales, empaquete el proyecto:
```
mvn clean install
```
Inicie sesión en su clúster EMR mediante SSH. Para obtener más información, consulte Log on to a cluster.
Cargue el paquete JAR en el directorio raíz del clúster EMR.

Paso 6: Envío del trabajo de Spark

Ejecute spark-submit para iniciar el trabajo de streaming:

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --driver-memory 1g \
  --executor-cores 2 \
  --executor-memory 3g \
  --num-executors 1 \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:<version> \
  --class com.aliyun.iceberg.StructuredSinkIceberg \
  iceberg-demos.jar

Nota

Reemplace <version> con la versión correspondiente. spark-sql-kafka debe ser compatible tanto con su versión de Spark como con la de Kafka.
Este ejemplo utiliza un paquete JAR llamado iceberg-demos.jar. Ajuste --class y el nombre del JAR para que coincidan con su proyecto.

Para comprobar que los datos se están escribiendo correctamente, utilice Spark SQL para consultar los cambios. Para obtener más información, consulte Basic usage.

E-MapReduce:Use Spark to write data to an Iceberg table in streaming mode