Informações de configuração do Delta Lake no EMR e exemplos de comandos comuns - E-MapReduce: Central de Documentação do Alibaba Cloud

O Delta Lake é pré-integrado ao E-MapReduce (EMR) e oferece transações ACID, gerenciamento escalável de metadados e processamento unificado de dados em lote e streaming para cargas de trabalho Apache Spark no armazenamento existente do data lake.

Informações de configuração

O Delta Lake é ativado no EMR por meio de extensões de sessão do Spark. A configuração necessária varia conforme a versão do Spark.

Spark 2.X

spark.sql.extensions  io.delta.sql.DeltaSparkSessionExtension

Spark 3.X

spark.sql.extensions        io.delta.sql.DeltaSparkSessionExtension
spark.sql.catalog.spark_catalog  org.apache.spark.sql.delta.catalog.DeltaCatalog

Essas configurações são aplicadas automaticamente ao adicionar o serviço Delta Lake ao cluster EMR. Para aplicá-las manualmente, use-as como flags --conf ao iniciar o streaming-sql.

Operações comuns

Todos os exemplos nesta seção usam uma tabela Delta chamada delta_table.

Crie uma tabela

CREATE TABLE delta_table (id INT) USING delta;

Inserir dados

INSERT INTO delta_table VALUES 0,1,2,3,4;

Verifique a inserção:

SELECT * FROM delta_table;

Sobrescrever dados

INSERT OVERWRITE TABLE delta_table VALUES 5,6,7,8,9;

Verifique o resultado:

SELECT * FROM delta_table;

Atualize dados

Some 100 a todos os IDs pares:

UPDATE delta_table SET id = id + 100 WHERE mod(id, 2) = 0;

Verifique a atualização:

SELECT * FROM delta_table;

Exclua dados

Exclua todos os registros com IDs pares:

DELETE FROM delta_table WHERE mod(id, 2) = 0;

Verifique a exclusão:

SELECT * FROM delta_table;

Mesclar dados

A instrução MERGE INTO executa upsert de dados de uma tabela de origem em uma tabela Delta. Use-a para aplicar atualizações e inserções condicionais em uma única operação atômica.

Crie uma tabela de origem para a operação de mesclagem.
```
CREATE TABLE newData(id INT) USING delta;
```

Insira dados na tabela de origem.

INSERT INTO newData VALUES 0,1,2,3,4,5,6,7,8,9;

Mescle os dados de newData em delta_table. Se o ID de um registro em newData corresponder a um ID em delta_table, some 100 ao ID correspondente. Caso contrário, insira o registro diretamente.
```
MERGE INTO delta_table AS target
  USING newData AS source
  ON target.id = source.id
  WHEN MATCHED THEN UPDATE SET target.id = source.id + 100
  WHEN NOT MATCHED THEN INSERT *;
```
Verifique o resultado da mesclagem.
```
SELECT * FROM delta_table;
```

Ler uma tabela Delta em modo streaming

Estes passos usam a CLI streaming-sql do EMR para configurar um pipeline de streaming a partir de uma tabela Delta existente.

O passo 6 exige uma segunda sessão do cliente streaming-sql em execução simultânea. Abra um segundo terminal antes de iniciar o passo 6.

Pré-requisitos

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

Um cluster EMR com acesso SSH
Uma tabela Delta existente chamada delta_table

Configurar o pipeline de streaming

Faça logon no cluster em modo SSH.
Inicie o cliente streaming-sql.
```
streaming-sql
```
Se o Delta Lake não estiver adicionado como serviço ao cluster, inicie o streaming-sql com o JAR e a configuração do Delta Lake: streaming-sql --jars /path/to/delta-core_2.11-0.6.1.jar --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension

Crie a tabela de destino.

CREATE TABLE stream_debug_table (id INT) USING DELTA;

Crie um scan de stream na tabela Delta de origem.

CREATE SCAN stream_delta_table on delta_table USING STREAM;

Inicie o job de streaming para gravar dados de delta_table em stream_debug_table.

CREATE STREAM job
options (
  triggerType='ProcessingTime',
  checkpointLocation = '/tmp/streaming_read_cp'
)
INSERT INTO stream_debug_table
SELECT * FROM stream_delta_table;

Verifique o comportamento do streaming

Em uma segunda sessão do cliente streaming-sql, consulte o estado atual da tabela de destino.
```
SELECT * FROM stream_debug_table;
```
Na segunda sessão, insira dados na tabela de origem.
```
INSERT INTO delta_table VALUES 801, 802;
```
Consulte novamente a tabela de destino para confirme que os novos dados foram transmitidos via streaming.
```
SELECT * FROM stream_debug_table;
```
Insira outro lote de dados na tabela de origem.
```
INSERT INTO delta_table VALUES 901, 902;
```
Consulte a tabela de destino para confirme que o segundo lote foi transmitido via streaming.
```
SELECT * FROM stream_debug_table;
```