Use Paimon tables in EMR Serverless Spark - E-MapReduce: Central de Documentação do Alibaba Cloud

Pré-requisitos

Procedimento

Etapa 1: Criar uma sessão SQL

Acesse a página Sessions.
1. Faça login no console EMR.
2. No painel de navegação à esquerda, escolha EMR Serverless > Spark.
3. Na página Spark, clique em nome do workspace desejado.
4. Na página EMR Serverless Spark, clique em Sessions no painel de navegação à esquerda.
Na página SQL Session, clique em Connect to SQL Session.

Na página Create SQL Session, configure os parâmetros na seção Spark Configuration e clique em Create. Gerenciar sessões SQL.

O Spark lê e grava tabelas Paimon por meio de catálogos. Selecione o tipo de catálogo conforme o seu cenário. Gerenciar catálogos de dados.

Catálogo de dados

Ao utilizar um catálogo de dados, não são necessários parâmetros de sessão. Clique em Catalog na página Add Catalog e selecione o catálogo de dados no ambiente de desenvolvimento Spark SQL.

Nota

Recomendamos as versões do mecanismo EMR esr-4.3.0 ou posterior, esr-3.3.0 ou posterior, ou esr-2.7.0 ou posterior.

Catálogo personalizado

Data Lake Formation (DLF)

spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       rest
spark.sql.catalog.<catalogName>.uri                             http://cn-hangzhou-vpc.dlf.aliyuncs.com
spark.sql.catalog.<catalogName>.warehouse                       <catalog_name>
spark.sql.catalog.<catalogName>.token.provider                  dlf
spark.sql.catalog.<catalogName>.dlf.access-key-id               <access_key_id>
spark.sql.catalog.<catalogName>.dlf.access-key-secret           <access_key_secret>

Parâmetros:

Parâmetro	Descrição	Valor de exemplo
`spark.sql.catalog.<catalogName>`	Implementação do catálogo.	Valor fixo: `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	Método de armazenamento de metadados. Defina como `rest` para usar a API REST do DLF.	Valor fixo: `rest`
`spark.sql.catalog.<catalogName>.uri`	URI do DLF. Formato: `http://<endpoint>-vpc.dlf.aliyuncs.com`.	`http://cn-hangzhou-vpc.dlf.aliyuncs.com`
`spark.sql.catalog.<catalogName>.warehouse`	Caminho do warehouse. Para o DLF, defina este valor como o nome do catálogo.	`<catalog_name>`
`spark.sql.catalog.<catalogName>.token.provider`	Provedor de autenticação. Defina como `dlf` para o DLF.	Valor fixo: `dlf`
`spark.sql.catalog.<catalogName>.dlf.access-key-id`	AccessKey ID da sua conta Alibaba Cloud ou usuário RAM.	`<access_key_id>`
`spark.sql.catalog.<catalogName>.dlf.access-key-secret`	AccessKey secret da sua conta Alibaba Cloud ou usuário RAM.	`<access_key_secret>`

DLF-Legacy

Os metadados são armazenados no DLF 1.0.

spark.sql.catalog.<catalogName>                          org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                dlf
spark.sql.catalog.<catalogName>.dlf.catalog.id           <catalog_name>
spark.sql.catalog.<catalogName>.dlf.catalog.endpoint     dlf-vpc.cn-hangzhou.aliyuncs.com

Parâmetros:

Parâmetro	Descrição	Valor de exemplo
`spark.sql.catalog.<catalogName>`	Implementação do catálogo.	Valor fixo: `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	Método de armazenamento de metadados. Defina como `dlf` para usar o DLF como metastore.	Valor fixo: `dlf`
`spark.sql.catalog.<catalogName>.dlf.catalog.id`	Nome do catálogo no DLF.	`<catalog_name>`
`spark.sql.catalog.<catalogName>.dlf.catalog.endpoint`	Endpoint do DLF. Selecione um endpoint do DLF com base na sua região.	`dlf-vpc.cn-hangzhou.aliyuncs.com`

Hive Metastore

Os metadados são armazenados em um Hive Metastore especificado.

spark.sql.catalog.<catalogName>                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore       hive
spark.sql.catalog.<catalogName>.uri             thrift://<yourHMSUri>:<port>

Parâmetros:

Parâmetro	Descrição	Valor de exemplo
`spark.sql.catalog.<catalogName>`	Implementação do catálogo.	Valor fixo: `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	Tipo de metastore. Defina como `hive` para usar um Hive Metastore.	Valor fixo: `hive`
`spark.sql.catalog.<catalogName>.uri`	URI do Hive Metastore. Formato: `thrift://<IP address of Hive metastore>:9083`. `<IP address of Hive metastore>` é o endereço IP interno do serviço Hive Metastore. Para usar um metastore externo, consulte Conectar a um serviço externo de Hive Metastore.	`thrift://192.168..:9083`

Sistema de arquivos

Os metadados são armazenados em um sistema de arquivos.

spark.sql.catalog.<catalogName>                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore       filesystem
spark.sql.catalog.<catalogName>.warehouse       oss://<yourBucketName>/warehouse

Parâmetros:

Parâmetro	Descrição	Valor de exemplo
`spark.sql.catalog.<catalogName>`	Implementação do catálogo.	Valor fixo: `org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	Tipo de metastore. Defina como `filesystem` para usar um sistema de arquivos como metastore.	Valor fixo: `filesystem`
`spark.sql.catalog.<catalogName>.warehouse`	Caminho do warehouse. Neste exemplo, `<yourBucketName>` é o nome de um bucket do OSS.	`oss://my-bucket/warehouse`

É possível configurar vários catálogos simultaneamente, como DLF, DLF 1.0 e Hive. Exemplo:

# Configure a DLF catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       rest
spark.sql.catalog.<catalogName>.uri                             http://cn-hangzhou-vpc.dlf.aliyuncs.com
spark.sql.catalog.<catalogName>.warehouse                       <catalog_name>
spark.sql.catalog.<catalogName>.token.provider                  dlf
spark.sql.catalog.<catalogName>.dlf.access-key-id               <access_key_id>
spark.sql.catalog.<catalogName>.dlf.access-key-secret           <access_key_secret>

# Configure a DLF 1.0 catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       dlf
spark.sql.catalog.<catalogName>.dlf.catalog.id                  <catalog_name>
spark.sql.catalog.<catalogName>.dlf.catalog.endpoint            dlf-vpc.cn-hangzhou.aliyuncs.com

# Configure a hive1 catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       hive
spark.sql.catalog.<catalogName>.uri                             thrift://<yourHMSUri-1>:<port>

# Configure a hive2 catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       hive
spark.sql.catalog.<catalogName>.uri                             thrift://<yourHMSUri-2>:<port>

Etapa 2: Ler e gravar tabelas Paimon

Acesse a página de desenvolvimento SQL.

Na página EMR Serverless Spark, clique em Development no painel de navegação à esquerda.
Na aba Development, clique em ícone .
Na caixa de diálogo Create, insira um nome como users_task, mantenha o Type como SparkSQL (padrão) e clique em OK.

Copie o código abaixo para a nova aba Spark SQL (users_task).

Catálogo Paimon

-- Create a database.
CREATE DATABASE IF NOT EXISTS paimon.ss_paimon_db;             

-- Create a Paimon table.
CREATE TABLE paimon.ss_paimon_db.paimon_tbl (id INT, name STRING) USING paimon;

-- Write data to the Paimon table.
INSERT INTO paimon.ss_paimon_db.paimon_tbl VALUES (1, "a"), (2, "b"), (3, "c");

-- Query the write results from the Paimon table.
SELECT * FROM paimon.ss_paimon_db.paimon_tbl ORDER BY id;

-- Delete the database.
DROP DATABASE paimon.ss_paimon_db CASCADE;

Selecione um banco de dados na lista suspensa de bancos de dados e a sessão SQL criada na lista suspensa de sessões.
Clique em Run para executar o job. O job retorna a seguinte saída:

Perguntas frequentes

Por que recebo um erro ao executar uma instrução DELETE, UPDATE ou MERGE em uma tabela?

Sintoma: A execução de uma instrução DELETE, UPDATE ou MERGE retorna um erro semelhante ao seguinte:

Caused by: org.apache.spark.sql.AnalysisException: Table does not support deletes/updates/merge: <tableName>.
    at org.apache.spark.sql.errors.QueryCompilationErrors$.tableDoesNotSupportError(QueryCompilationErrors.scala:1391)

Causa: O formato de armazenamento da tabela não oferece suporte a atualizações no nível de linha ou falta uma configuração necessária do Spark.
Solução:
1. Verifique o tipo da tabela.
  
  Execute o comando a seguir para confirmar se a tabela é uma tabela Paimon:
```
SHOW CREATE TABLE <tableName>;
```
  Se a saída contiver USING PAIMON, trata-se de uma tabela Paimon. Caso a saída indique outro formato de armazenamento (como USING hive), verifique se esse formato suporta atualizações no nível de linha.
2. Valide a configuração do Spark.
  
  Se for uma tabela Paimon, verifique a seção Spark Configuration e garanta que a seguinte configuração esteja presente:
```
spark.sql.extensions org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
```
  Caso esteja ausente, adicione-a em Spark Configuration.

Documentação relacionada

Início rápido para desenvolvimento SparkSQL
Para detalhes sobre uso e configuração do Paimon, consulte a documentação oficial do Paimon.
Para usar um serviço de metastore externo, consulte Conectar a um serviço externo de Hive Metastore.