Suporte ao formato de tabela aberta

Configuração do Spark para Iceberg: Se você quiser usar o formato Iceberg, defina as seguintes configurações. Substitua DB_LOCATION pelo caminho do Amazon S3 onde suas tabelas do Iceberg estão localizadas e os espaços reservados para a região e o ID da conta por seus próprios valores.

spark-sql \
--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
--conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog 
--conf spark.sql.catalog.spark_catalog.warehouse=s3://DB_LOCATION
--conf spark.sql.catalog.spark_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog 
--conf spark.sql.catalog.spark_catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO
--conf spark.sql.catalog.spark_catalog.glue.account-id=ACCOUNT_ID
--conf spark.sql.catalog.spark_catalog.glue.id=ACCOUNT_ID
--conf spark.sql.catalog.spark_catalog.client.region=AWS_REGION
		

Se quiser usar o formato do Iceberg em versões anteriores do EMR, use o seguinte comando:

spark-sql \
--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,com.amazonaws.emr.recordserver.connector.spark.sql.RecordServerSQLExtension  
--conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkCatalog 
--conf spark.sql.catalog.spark_catalog.warehouse=s3://DB_LOCATION
--conf spark.sql.catalog.spark_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog 
--conf spark.sql.catalog.spark_catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO  
--conf spark.sql.catalog.spark_catalog.glue.account-id=ACCOUNT_ID
--conf spark.sql.catalog.spark_catalog.glue.id=ACCOUNT_ID
--conf spark.sql.catalog.spark_catalog.client.assume-role.region=AWS_REGION
--conf spark.sql.catalog.spark_catalog.lf.managed=true
		

Exemplos:

Aqui estão alguns exemplos de como trabalhar com tabelas Iceberg:

-- Create an Iceberg table
CREATE TABLE my_iceberg_table (
    id BIGINT,
    name STRING,
    created_at TIMESTAMP
) USING ICEBERG;

-- Insert data
INSERT INTO my_iceberg_table VALUES (1, 'Alice', current_timestamp());

-- Query the table
SELECT * FROM my_iceberg_table;

Configuração do Spark para Hudi:

Para iniciar o shell Spark no EMR 7.10 ou versões superiores, use o comando a seguir:

spark-sql
--jars /usr/lib/hudi/hudi-spark-bundle.jar \
--conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
--conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension

Para iniciar o shell Spark em versões anteriores do EMR, use o comando abaixo em vez disso:

spark-sql
--jars /usr/lib/hudi/hudi-spark-bundle.jar \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
--conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
--conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension,com.amazonaws.emr.recordserver.connector.spark.sql.RecordServerSQLExtension  \
--conf spark.sql.catalog.spark_catalog.lf.managed=true  

Exemplos:

Aqui estão alguns exemplos de como trabalhar com tabelas Hudi:

-- Create a Hudi table
CREATE TABLE my_hudi_table (
    id BIGINT,
    name STRING,
    created_at TIMESTAMP
) USING HUDI
TBLPROPERTIES (
    'type' = 'cow',
    'primaryKey' = 'id'
);

-- Insert data
INSERT INTO my_hudi_table VALUES (1, 'Alice', current_timestamp());

-- Query the latest snapshot
SELECT * FROM my_hudi_table;

Para consultar o último instantâneo das copy-on-write tabelas:

SELECT * FROM my_hudi_cow_table

spark.read.table("my_hudi_cow_table")

Para consultar os dados compactados mais recentes das tabelas MOR, você pode consultar a tabela otimizada para leitura que tem o sufixo _ro:

SELECT * FROM my_hudi_mor_table_ro

spark.read.table("my_hudi_mor_table_ro")

Configuração do Spark para Delta Lake:

Para usar o Delta Lake com o Lake Formation no EMR 7.10 e superior, execute o seguinte comando:

spark-sql \
   --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog

Para usar o Delta Lake com o Lake Formation no EMR 6.15 a 7.9, execute o seguinte

spark-sql \
  --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension,com.amazonaws.emr.recordserver.connector.spark.sql.RecordServerSQLExtension \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
  --conf spark.sql.catalog.spark_catalog.lf.managed=true

Se quiser que o Lake Formation use o servidor de registros para gerenciar o catálogo do Spark, defina spark.sql.catalog.<managed_catalog_name>.lf.managed como verdadeiro.

Exemplos:

Aqui estão alguns exemplos de como trabalhar com tabelas Delta Lake:

-- Create a Delta Lake table
CREATE TABLE my_delta_table (
    id BIGINT,
    name STRING,
    created_at TIMESTAMP
) USING DELTA;

-- Insert data
INSERT INTO my_delta_table VALUES (1, 'Alice', current_timestamp());

-- Query the table
SELECT * FROM my_delta_table;

-- Update data
UPDATE my_delta_table SET name = 'Alice Smith' WHERE id = 1;

-- Merge data
MERGE INTO my_delta_table AS target
USING (SELECT 2 as id, 'Bob' as name, current_timestamp() as created_at) AS source
ON target.id = source.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

Criação de uma tabela Delta Lake no AWS Glue Data Catalog

O Amazon EMR com Lake Formation não oferece suporte a comandos DDL e à criação de tabelas Delta em versões do EMR anteriores à 7.12. Siga estas etapas para criar tabelas no AWS Glue Data Catalog.