Write data to an Iceberg table by using Spark Structured Streaming. - E-MapReduce: Central de Documentação do Alibaba Cloud

Use o Apache Spark Structured Streaming para gravar continuamente dados do Apache Kafka em uma tabela do Apache Iceberg no EMR. A API DataStreamWriter gerencia todas as gravações.

Pré-requisitos

Antes de começar, verifique se você tem:

Um cluster DataLake ou personalizado. Para mais informações, consulte Criar um cluster.
Um cluster Dataflow com o serviço Kafka ativado. Para mais informações, consulte Criar um cluster.
Ambos os clusters implantados no mesmo vSwitch da mesma Virtual Private Cloud (VPC).

Como funciona

Envie um job de gravação em streaming chamando a API DataStreamWriter do 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()

O parâmetro tableIdentifier representa o nome ou o caminho da tabela de metadados de destino. Há suporte para dois modos de saída:

Modo	Comportamento	Equivalente SQL
`append`	Grava cada microlote como novas linhas	`INSERT INTO`
`complete`	Substitui toda a tabela pelo microlote mais recente	`INSERT OVERWRITE`

Transmitir dados do Kafka para uma tabela do Iceberg

Este exemplo lê dados de um cluster Dataflow (Kafka) e os grava em uma tabela do Iceberg. Após empacotar o código, envie-o como um job do Spark usando o comando spark-submit.

Etapa 1: Configurar um tópico do Kafka com dados de teste

Faça login no cluster Dataflow via SSH. Para mais informações, consulte Fazer login em um cluster.

Crie um tópico chamado 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

Produza dados de teste para o tópico:

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

Etapa 2: Criar um banco de dados e uma tabela no Iceberg

Use o Spark SQL para criar um banco de dados chamado iceberg_db e uma tabela chamada iceberg_table. Para mais informações, consulte Usar o Iceberg.

Etapa 3: Configurar o projeto Maven

Crie um projeto Maven, adicione as dependências do Spark e inclua o plugin Scala Maven para compilação. Adicione o seguinte conteúdo ao arquivo 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>

Etapa 4: Escrever a aplicação Spark

O código Scala a seguir lê dados do Kafka e os grava em uma tabela do Iceberg no modo de streaming.

Importante

Os parâmetros de catálogo e os nomes padrão variam conforme a versão do cluster. Este exemplo usa um cluster EMR V5.3.0 com o DLF (Data Lake Formation) gerenciando metadados sob um catálogo chamado dlf_catalog. Para obter detalhes sobre a configuração do DLF, consulte Configuração de metadados do DLF.

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()
}

Ajuste os valores dos parâmetros descritos na tabela a seguir conforme suas necessidades de negócio.

Variável	Descrição
`checkpointPath`	Caminho no OSS para os dados de checkpoint do Spark Structured Streaming
`bootstrapServers`	Endereço IP privado de um broker Kafka no cluster Dataflow
`topic`	Nome do tópico do Kafka para assinatura

Etapa 5: Empacotar e implantar no EMR

Após testar localmente, empacote o projeto:
```
mvn clean install
```
Faça login no cluster EMR via SSH. Para mais informações, consulte Fazer login em um cluster.
Carregue o pacote JAR no diretório raiz do cluster EMR.

Etapa 6: Enviar o job do Spark

Execute o comando spark-submit para iniciar o job 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

Substitua <version> pela versão real. A biblioteca spark-sql-kafka deve ser compatível com as versões do Spark e do Kafka.
Este exemplo usa um pacote JAR chamado iceberg-demos.jar. Ajuste o parâmetro --class e o nome do JAR para corresponderem ao seu projeto.

Para verificar se a gravação dos dados está ocorrendo corretamente, use o Spark SQL para consultar as alterações. Para mais informações, consulte Uso básico.

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