Query AWS Glue Catalog in a Notebook

import re
from sedona.spark import *
from pyspark.sql import functions as F

# ----------------------------------------------------------------------
# Customer configuration
# ----------------------------------------------------------------------

CATALOG_NAME = "YOUR_CATALOG_NAME"   # Glue catalog name from Data Hub
DB_NAME = "YOUR_DB_NAME"                # Database (namespace) name
TABLE_NAME = "YOUR_TABLE_NAME"

SOURCE_FILE = "s3://YOUR_BUCKET/YOUR_FILE.csv"

# S3 prefix where the Iceberg table data and metadata will be stored.
# This must be a prefix/folder, not the CSV file.
WAREHOUSE = "s3://YOUR_BUCKET/SUBDIRECTORY/"

OUT_TABLE = f"{CATALOG_NAME}.{DB_NAME}.{TABLE_NAME}"
TABLE_LOCATION = f"{WAREHOUSE.rstrip('/')}/{DB_NAME}/{TABLE_NAME}/"

# ----------------------------------------------------------------------
# Create Sedona / Spark session and select the namespace
# ----------------------------------------------------------------------
# Your Glue catalog is registered by Wherobots when you connect it in Data
# Hub and start a new runtime, so no catalog config is needed here. Use the
# fully qualified CATALOG_NAME.DB_NAME so statements route to the Glue catalog.

config = SedonaContext.builder().getOrCreate()
sedona = SedonaContext.create(config)

sedona.sql(f"CREATE NAMESPACE IF NOT EXISTS {CATALOG_NAME}.{DB_NAME}")
sedona.sql(f"USE {CATALOG_NAME}.{DB_NAME}")

    # --------------------------------------------
    # Read source CSV
    # --------------------------------------------

    raw_df = (
        sedona.read
            .option("header", True)
            .option("inferSchema", True)
            .csv(SOURCE_FILE)
    )

    # --------------------------------------------
    # Basic ETL: normalize column names, trim text,
    # and drop duplicates
    # --------------------------------------------

    def normalize_col_name(name: str) -> str:
        cleaned = re.sub(r"[^A-Za-z0-9_]", "_", name.strip().lower())
        cleaned = re.sub(r"_+", "_", cleaned).strip("_")
        return cleaned or "col"


    def make_unique_col_names(cols):
        seen = {}
        output = []

        for col in cols:
            base = normalize_col_name(col)
            count = seen.get(base, 0)

            if count == 0:
                output.append(base)
            else:
                output.append(f"{base}_{count + 1}")

            seen[base] = count + 1

        return output

    renamed_df = raw_df.toDF(*make_unique_col_names(raw_df.columns))

    string_cols = {
        col_name
        for col_name, dtype_name in renamed_df.dtypes
        if dtype_name == "string"
    }

    clean_df = (
        renamed_df
            .select(
                *[
                    F.trim(F.col(c)).alias(c) if c in string_cols else F.col(c)
                    for c in renamed_df.columns
                ]
            )
            .dropDuplicates()
    )

    # --------------------------------------------
    # Add synthetic foot traffic score
    # --------------------------------------------

    score_cols = [
        F.coalesce(F.col(c).cast("string"), F.lit(""))
        for c in clean_df.columns
    ]

    scored_df = (
        clean_df
            .withColumn(
                "synthetic_foot_traffic_score",
                (
                    F.pmod(F.xxhash64(*score_cols), F.lit(100))
                    + F.lit(1)
                ).cast("int")
            )
            .withColumn("_source_file", F.lit(SOURCE_FILE))
            .withColumn("_ingested_at", F.current_timestamp())
    )

    # --------------------------------------------
    # Create Iceberg table, then insert data
    # --------------------------------------------

    def quote_identifier(name: str) -> str:
        return f"`{name.replace('`', '``')}`"


    columns_sql = ",\n  ".join(
        f"{quote_identifier(field.name)} {field.dataType.simpleString()}"
        for field in scored_df.schema.fields
    )

    select_sql = ",\n  ".join(
        quote_identifier(field.name)
        for field in scored_df.schema.fields
    )

    scored_df.createOrReplaceTempView("scored_source_data")

    sedona.sql(f"DROP TABLE IF EXISTS {OUT_TABLE}")

    sedona.sql(f"""
    CREATE TABLE {OUT_TABLE} (
    {columns_sql}
    )
    USING iceberg
    LOCATION '{TABLE_LOCATION}'
    TBLPROPERTIES (
    'format-version' = '2'
    )
    """)

    sedona.sql(f"""
    INSERT INTO {OUT_TABLE}
    SELECT
    {select_sql}
    FROM scored_source_data
    """)

    # --------------------------------------------
    # Verify table
    # --------------------------------------------

    sedona.sql(f"SELECT * FROM {OUT_TABLE} LIMIT 10").show(truncate=False)

    sedona.sql(f"""
    SELECT
    file_format,
    COUNT(*) AS file_count
    FROM {OUT_TABLE}.files
    GROUP BY file_format
    """).show(truncate=False)