Configuration information of Delta Lake in EMR and examples of common commands - E-MapReduce

Delta Lake está preintegrado en E-MapReduce (EMR) y proporciona a sus cargas de trabajo de Apache Spark transacciones ACID, gestión escalable de metadatos y procesamiento unificado de datos por lotes y en streaming sobre su almacenamiento de data lake existente.

Información de configuración

Delta Lake se habilita en EMR mediante extensiones de sesión de Spark. La configuración obligatoria varía según la versión de 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

Estos parámetros se aplican de forma predeterminada al agregar el servicio Delta Lake a su clúster de EMR. Para aplicarlos manualmente, páselos como indicadores --conf al iniciar streaming-sql.

Operaciones comunes

Todos los ejemplos de esta sección utilizan una tabla Delta llamada delta_table.

Crear una tabla

CREATE TABLE delta_table (id INT) USING delta;

Insertar datos

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

Compruebe la inserción:

SELECT * FROM delta_table;

Sobrescribir datos

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

Compruebe el resultado:

SELECT * FROM delta_table;

Actualizar datos

Sume 100 a todos los ID pares:

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

Compruebe la actualización:

SELECT * FROM delta_table;

Eliminar datos

Elimine todos los registros con ID pares:

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

Compruebe la eliminación:

SELECT * FROM delta_table;

Combinar datos

La instrucción MERGE INTO inserta o actualiza datos desde una tabla de origen hacia una tabla Delta. Utilícela para aplicar actualizaciones e inserciones condicionales en una sola operación atómica.

Cree una tabla de origen para la operación de combinación.
```
CREATE TABLE newData(id INT) USING delta;
```

Inserte datos en la tabla de origen.

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

Combine los datos de newData en delta_table. Si el ID de un registro en newData coincide con un ID en delta_table, sume 100 al ID coincidente. De lo contrario, inserte el registro directamente.
```
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 *;
```
Compruebe el resultado de la combinación.
```
SELECT * FROM delta_table;
```

Leer una tabla Delta en modo streaming

Estos pasos utilizan la CLI streaming-sql de EMR para configurar una canalización de streaming a partir de una tabla Delta existente.

El paso 6 requiere una segunda sesión del cliente streaming-sql que se ejecute simultáneamente. Abra una segunda terminal antes de comenzar el paso 6.

Requisitos previos

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

Un clúster de EMR con acceso SSH
Una tabla Delta existente llamada delta_table

Configurar la canalización de streaming

Log on to your cluster in SSH mode.
Inicie el cliente streaming-sql.
```
streaming-sql
```
Si Delta Lake no está agregado como servicio en su clúster, inicie streaming-sql con el JAR de Delta Lake y la configuración correspondiente: streaming-sql --jars /path/to/delta-core_2.11-0.6.1.jar --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension

Cree la tabla de destino.

CREATE TABLE stream_debug_table (id INT) USING DELTA;

Cree un escaneo de flujo sobre la tabla Delta de origen.

CREATE SCAN stream_delta_table on delta_table USING STREAM;

Inicie el trabajo de streaming para escribir datos desde delta_table hacia stream_debug_table.

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

Verificar el comportamiento del streaming

En una segunda sesión del cliente streaming-sql, consulte el estado actual de la tabla de destino.
```
SELECT * FROM stream_debug_table;
```
En la segunda sesión, inserte datos en la tabla de origen.
```
INSERT INTO delta_table VALUES 801, 802;
```
Consulte nuevamente la tabla de destino para confirmar que los nuevos datos se transmitieron correctamente.
```
SELECT * FROM stream_debug_table;
```
Inserte otro lote de datos en la tabla de origen.
```
INSERT INTO delta_table VALUES 901, 902;
```
Consulte la tabla de destino para confirmar que el segundo lote se transmitió correctamente.
```
SELECT * FROM stream_debug_table;
```