Prise en charge du format de tableau ouvert

Configuration Spark pour Iceberg : si vous souhaitez utiliser le format Iceberg, définissez les configurations suivantes. DB_LOCATIONRemplacez-le par le chemin Amazon S3 où se trouvent vos tables Iceberg, et remplacez les espaces réservés à la région et à l'ID de compte par vos propres valeurs.

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
		

Si vous souhaitez utiliser le format Iceberg sur les versions antérieures d'EMR, utilisez plutôt la commande suivante :

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
		

Exemples :

Voici quelques exemples d'utilisation des tables 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;

Configuration de Spark pour Hudi :

Pour démarrer le shell Spark sur EMR 7.10 ou version ultérieure, utilisez la commande suivante :

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

Pour démarrer le shell Spark sur des versions antérieures d'EMR, utilisez plutôt la commande ci-dessous :

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  

Exemples :

Voici quelques exemples d'utilisation des tables 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;

Pour demander le dernier instantané des copy-on-write tables :

SELECT * FROM my_hudi_cow_table

spark.read.table("my_hudi_cow_table")

Pour interroger les dernières données compactées des tables MOR, vous pouvez interroger la table optimisée en lecture qui est suffixée par _ro :

SELECT * FROM my_hudi_mor_table_ro

spark.read.table("my_hudi_mor_table_ro")

Configuration de Spark pour Delta Lake :

Pour utiliser Delta Lake with Lake Formation sur EMR 7.10 ou version ultérieure, exécutez la commande suivante :

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

Pour utiliser le lac Delta avec Lake Formation sur EMR 6,15 à 7,9, procédez comme suit

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

Si vous souhaitez que Lake Formation utilise un serveur d'enregistrement pour gérer votre catalogue Spark, définissez spark.sql.catalog.<managed_catalog_name>.lf.managed ce paramètre sur true.

Exemples :

Voici quelques exemples d'utilisation des tables 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 *;

Création d'une table Delta Lake dans AWS Glue Data Catalog

Amazon EMR with Lake Formation ne prend pas en charge les commandes DDL ni la création de tables Delta dans les versions d'EMR antérieures à 7.12. Procédez comme suit pour créer des tables dans le catalogue de données AWS Glue.