Unterstützung für das offene Tabellenformat

Spark-Konfiguration für Iceberg: Wenn Sie das Iceberg-Format verwenden möchten, legen Sie die folgenden Konfigurationen fest. DB_LOCATIONErsetzen Sie durch den Amazon S3 S3-Pfad, in dem sich Ihre Iceberg-Tabellen befinden, und ersetzen Sie die Platzhalter für Region und Konto-ID durch Ihre eigenen Werte.

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
		

Wenn Sie das Iceberg-Format in früheren EMR-Versionen verwenden möchten, verwenden Sie stattdessen den folgenden Befehl:

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
		

Beispiele:

Hier sind einige Beispiele für die Arbeit mit Iceberg-Tabellen:

-- 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;

Spark-Konfiguration für Hudi:

Verwenden Sie den folgenden Befehl, um die Spark-Shell auf EMR 7.10 oder höheren Versionen zu starten:

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

Um die Spark-Shell auf früheren EMR-Versionen zu starten, verwenden Sie stattdessen den folgenden Befehl:

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  

Beispiele:

Hier sind einige Beispiele für die Arbeit mit Hudi-Tabellen:

-- 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;

Um den neuesten Snapshot von copy-on-write Tabellen abzufragen:

SELECT * FROM my_hudi_cow_table

spark.read.table("my_hudi_cow_table")

Um die neuesten komprimierten Daten von MOR-Tabellen abzufragen, können Sie die leseoptimierte Tabelle mit dem Suffix _ro abfragen:

SELECT * FROM my_hudi_mor_table_ro

spark.read.table("my_hudi_mor_table_ro")

Spark-Konfiguration für Delta Lake:

Führen Sie den folgenden Befehl aus, um Delta Lake mit Lake Formation auf EMR 7.10 und höher zu verwenden:

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

Um Delta Lake mit Lake Formation auf EMR 6.15 bis 7.9 zu verwenden, führen Sie den folgenden Befehl aus

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

Wenn Sie möchten, dass Lake Formation einen Datensatzserver zur Verwaltung Ihres Spark-Katalogs verwendet, legen Sie spark.sql.catalog.<managed_catalog_name>.lf.managed den Wert auf true fest.

Beispiele:

Hier sind einige Beispiele für die Arbeit mit Delta Lake-Tabellen:

-- 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 *;

Erstellen einer Delta Lake-Tabelle im AWS Glue Data Catalog

Amazon EMR with Lake Formation unterstützt keine DDL-Befehle und die Erstellung von Delta-Tabellen in EMR-Versionen vor 7.12. Gehen Sie wie folgt vor, um Tabellen im AWS Glue-Datenkatalog zu erstellen.