Use Iceberg in EMR - E-MapReduce: Central de Documentação do Alibaba Cloud

O Apache Iceberg no EMR permite executar operações INSERT, SELECT, UPDATE e DELETE em dados com formato de tabela armazenados no OSS, com metadados gerenciados pelo Data Lake Formation (DLF). Este tópico orienta você na configuração de um catálogo Iceberg integrado ao DLF e na execução de operações básicas do Spark SQL em um cluster Hadoop.

Pré-requisitos

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

Um cluster Hadoop executando o EMR V5.3.0 ou posterior. Consulte Criar um cluster
Acesso SSH ao cluster. Consulte Fazer logon em um cluster

Limitações

As extensões Spark SQL do Iceberg não são compatíveis com o Spark 2.4. Em clusters que executam o EMR V3.38.X ou posterior, use a API Spark DataFrame em vez do Spark SQL para interagir com o Iceberg. Este tópico aborda a abordagem Spark SQL para o EMR V5.3.0 e versões posteriores.

Configurar o catálogo Iceberg

Antes de executar o Spark SQL no Iceberg, configure um catálogo que aponte para o DLF como armazenamento de metadados. O nome do catálogo e os parâmetros necessários variam conforme a versão do cluster. Todas as configurações usam a mesma extensão Spark SQL:

org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

A configuração do catálogo usa o prefixo spark.sql.catalog.<catalog_name>, onde <catalog_name> é o nome atribuído ao catálogo.

Resumo das versões:

Versão do cluster	Nome padrão do catálogo	Classe `catalog-impl`
EMR V5.6.0 ou 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 ou posterior

Faça logon no cluster via SSH e execute:

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

Faça logon no cluster via SSH e execute:

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

Você pode deixar spark.sql.catalog.dlf.warehouse em branco. Se omitido, o sistema usará o caminho padrão do warehouse.

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

Esta versão exige um par de AccessKey. Defina as variáveis de ambiente ALIBABA_CLOUD_ACCESS_KEY_ID e ALIBABA_CLOUD_ACCESS_KEY_SECRET antes de executar o comando. Para obter instruções, consulte Configurar variáveis de ambiente no Linux, macOS e Windows.

Faça logon no cluster via SSH e execute:

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>

Substitua os espaços reservados pelos valores reais:

Espaço reservado	Descrição
`<your-oss-endpoint>`	Endpoint do OSS para sua região, por exemplo, `oss-cn-hangzhou.aliyuncs.com`
`<your-oss-warehouse-path>`	Caminho do OSS onde o Iceberg armazena os dados da tabela, por exemplo, `oss://my-bucket/warehouse`
`<ALIBABA_CLOUD_ACCESS_KEY_ID>`	Valor da variável de ambiente `ALIBABA_CLOUD_ACCESS_KEY_ID`
`<ALIBABA_CLOUD_ACCESS_KEY_SECRET>`	Valor da variável de ambiente `ALIBABA_CLOUD_ACCESS_KEY_SECRET`
`<your-catalog-id>`	ID do catálogo DLF
`<your-dlf-endpoint>`	Endpoint do serviço DLF
`<your-dlf-region-id>`	ID da região onde o DLF está implantado, por exemplo, `cn-hangzhou`

Verificar a sessão

Quando a CLI do Spark SQL estiver pronta, o prompt mudará para:

spark-sql>

Executar operações básicas do Iceberg

Os exemplos abaixo usam <catalog_name> para referenciar o nome do catálogo configurado. Para o EMR V5.6.0 e posterior, substitua <catalog_name> por iceberg. Para outras versões, use o nome do catálogo indicado na seção Configurar o catálogo Iceberg.

Criar um banco de dados

CREATE DATABASE IF NOT EXISTS <catalog_name>.iceberg_db;

Criar uma tabela

A forma básica especifica nomes de colunas, tipos e o mecanismo de armazenamento do Iceberg:

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

O comando CREATE TABLE também aceita as cláusulas COMMENT, PARTITIONED BY, LOCATION e TBLPROPERTIES. Use TBLPROPERTIES para definir propriedades no nível da tabela, como o formato de gravação padrão:

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

Inserir dados

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

Consultar dados

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

Atualizar dados

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

Excluir dados

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

Tópicos relacionados

Para detalhes sobre a configuração de metadados, consulte Configuração de metadados do DLF.