Native Apache Iceberg Support in AWS Glue: What You Need to Know (and Probably Missed)

				
					import sys
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.context import SparkContext
from datetime import datetime
from pprint import pprint

# Get job args
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

# Set up Spark and Glue context
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Config
glue_database = "sales_database"
parquet_table = "sales_table_parquet"
iceberg_table = "sales_table_iceberg"

# Iceberg catalog config
spark.conf.set("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog")
spark.conf.set("spark.sql.catalog.glue_catalog.warehouse", "s3://icebergdemo-s3/iceberg_data/")
spark.conf.set("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog")
spark.conf.set("spark.sql.catalog.glue_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")

# ------------ PARQUET READ -------------------
start_parquet = datetime.now()
print(f"\n⏱️ Start loading Parquet: {start_parquet}")

parquet_sql = f"""
                SELECT country, SUM(quantity * unit_price) AS total_sales
                FROM {glue_database}.{parquet_table}
                WHERE sale_date BETWEEN  date('2023-07-01') AND date('2023-07-31')
                GROUP BY country
                ORDER BY total_sales DESC
                """
iceberg_sql = f""" 
                SELECT country, SUM(quantity * unit_price) AS total_sales
                FROM {glue_database}.{iceberg_table}
                WHERE sale_date BETWEEN  date('2023-07-01') AND date('2023-07-31')
                GROUP BY country
                ORDER BY total_sales DESC
                """

print(f"\nThe parquet query: {parquet_sql}\n")
print(f"\nThe iceberg query: {iceberg_sql}\n")

df_parquet = spark.sql(parquet_sql)
pprint(df_parquet.collect())
#df_parquet.count()  # Trigger execution

end_parquet = datetime.now()
print(f"✅ Finished loading Parquet: {end_parquet}")
print(f"🕒 Time taken for Parquet load: {(end_parquet - start_parquet).total_seconds()} seconds\n")

# ------------ ICEBERG READ -------------------
start_iceberg = datetime.now()

print(f"\n⏱️ Start loading Iceberg: {start_iceberg}")

df_iceberg = spark.sql(iceberg_sql)
pprint(df_parquet.collect())
#df_iceberg.count()  # Trigger execution

end_iceberg = datetime.now()
print(f"✅ Finished loading Iceberg: {end_iceberg}")
print(f"🕒 Time taken for Iceberg load: {(end_iceberg - start_iceberg).total_seconds()} seconds\n")

job.commit()