Ausführung von ETL-Jobs in Amazon-S3-Tabellen mit AWS Glue

Voraussetzungen, die erforderlich sind, um die AWS-Analytikintegration von S3 Tables zur Ausführung von AWS Glue Jobs zu verwenden.

Voraussetzungen für die Verwendung des Amazon-S3-Tables-Iceberg REST-Endpunkts zur Ausführung von AWS Glue ETL-Jobs.

Voraussetzungen: Verwenden Sie den Amazon S3 Tables Catalog für Apache Iceberg, um AWS Glue-ETL-Aufträge auszuführen.

Als Voraussetzung für das Abfragen von Tabellen mit oder Spark unter AWS Glue Verwendung der AWS Analytics Services-Integration müssen Sie Ihre Tabellen-Buckets in AWS Analytics-Services integrieren

Sie können die Verbindung zu Ihrem Tabellen-Bucket über eine Spark-Sitzung in einem Job oder mit AWS Glue Studio-Magics in einer interaktiven Sitzung konfigurieren. Um die folgenden Beispiele zu verwenden, ersetzen Sie die Platzhalterwerte durch die Informationen für Ihren eigenen Tabellen-Bucket.

Sie können die Verbindung zu Ihrem Tabellen-Bucket über eine Spark-Sitzung in einem Job oder mit AWS Glue Studio-Magics in einer interaktiven Sitzung konfigurieren. Um die folgenden Beispiele zu verwenden, ersetzen Sie die Platzhalterwerte durch die Informationen für Ihren eigenen Tabellen-Bucket.

Um eine Verbindung zu Tabellen über den Amazon S3 Tables Catalog für Apache Iceberg herstellen zu können, müssen Sie zunächst die neueste Katalog-JAR-Datei herunterladen und in einen S3-Bucket hochladen. Wenn Sie dann Ihren Job erstellen, fügen Sie den Pfad zum Client-Katalog JAR als speziellen Parameter hinzu. Weitere Informationen zu Auftrags-Parametern in AWS Glue finden Sie im AWS Glue-Entwicklerhandbuch unter Spezielle Parameter, die in AWS Glue-Aufträgen verwendet werden.

Sie können die Verbindung zu Ihrem Tabellen-Bucket über eine Spark-Sitzung in einem Job oder mit AWS Glue Studio-Magics in einer interaktiven Sitzung konfigurieren. Um die folgenden Beispiele zu verwenden, ersetzen Sie die Platzhalterwerte durch die Informationen für Ihren eigenen Tabellen-Bucket.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SparkIcebergSQL") \
    .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.4.2") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.defaultCatalog","s3tables")
    .config("spark.sql.catalog.s3tables", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.s3tables.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog") \
    .config("spark.sql.catalog.s3tables.glue.id", "111122223333:s3tablescatalog/amzn-s3-demo-table-bucket") \
    .config("spark.sql.catalog.s3tables.warehouse", "s3://amzn-s3-demo-table-bucket/bucket/amzn-s3-demo-table-bucket") \
    .getOrCreate()

namespace = "new_namespace"
table = "new_table"

spark.sql("SHOW DATABASES").show()

spark.sql(f"DESCRIBE NAMESPACE {namespace}").show()

spark.sql(f"""
    CREATE TABLE IF NOT EXISTS {namespace}.{table} (
       id INT,
       name STRING,
       value INT
    )
""")

spark.sql(f"""
    INSERT INTO {namespace}.{table}
    VALUES 
       (1, 'ABC', 100),
       (2, 'XYZ', 200)
""")

spark.sql(f"SELECT * FROM {namespace}.{table} LIMIT 10").show()

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("glue-s3-tables-rest") \
    .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.4.2") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.defaultCatalog", "s3_rest_catalog") \
    .config("spark.sql.catalog.s3_rest_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.s3_rest_catalog.type", "rest") \
    .config("spark.sql.catalog.s3_rest_catalog.uri", "https://s3tables.Region.amazonaws.com/iceberg") \
    .config("spark.sql.catalog.s3_rest_catalog.warehouse", "arn:aws:s3tables:Region:111122223333:bucket/amzn-s3-demo-table-bucket") \
    .config("spark.sql.catalog.s3_rest_catalog.rest.sigv4-enabled", "true") \
    .config("spark.sql.catalog.s3_rest_catalog.rest.signing-name", "s3tables") \
    .config("spark.sql.catalog.s3_rest_catalog.rest.signing-region", "Region") \
    .config('spark.sql.catalog.s3_rest_catalog.io-impl','org.apache.iceberg.aws.s3.S3FileIO') \
    .config('spark.sql.catalog.s3_rest_catalog.rest-metrics-reporting-enabled','false') \
    .getOrCreate()

namespace = "s3_tables_rest_namespace"
table = "new_table_s3_rest"

spark.sql("SHOW DATABASES").show()

spark.sql(f"DESCRIBE NAMESPACE {namespace}").show()

spark.sql(f"""
    CREATE TABLE IF NOT EXISTS {namespace}.{table} (
       id INT,
       name STRING,
       value INT
    )
""")

spark.sql(f"""
    INSERT INTO {namespace}.{table}
    VALUES 
       (1, 'ABC', 100),
       (2, 'XYZ', 200)
""")

spark.sql(f"SELECT * FROM {namespace}.{table} LIMIT 10").show()

            

from pyspark.sql import SparkSession

#Spark session configurations
spark = SparkSession.builder.appName("glue-s3-tables") \
    .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.4.2") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.defaultCatalog", "s3tablesbucket") \
    .config("spark.sql.catalog.s3tablesbucket", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.s3tablesbucket.catalog-impl", "software.amazon.s3tables.iceberg.S3TablesCatalog") \
    .config("spark.sql.catalog.s3tablesbucket.warehouse", "arn:aws:s3tables:Region:111122223333:bucket/amzn-s3-demo-table-bucket") \
    .getOrCreate()

#Script
namespace = "new_namespace"
table = "new_table"

spark.sql(f"CREATE NAMESPACE IF NOT EXISTS s3tablesbucket.{namespace}")
spark.sql(f"DESCRIBE NAMESPACE {namespace}").show()

spark.sql(f"""
    CREATE TABLE IF NOT EXISTS {namespace}.{table} (
       id INT,
       name STRING,
       value INT
    )
""")

spark.sql(f"""
    INSERT INTO {namespace}.{table}
    VALUES 
       (1, 'ABC', 100),
       (2, 'XYZ', 200)
""")

spark.sql(f"SELECT * FROM {namespace}.{table} LIMIT 10").show()