Query Amazon S3 Tables Catalog in a Notebook

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

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

CATALOG_NAME = "YOUR_CATALOG_NAME"   # S3 Tables catalog name from Data Hub
DB_NAME = "YOUR_DB_NAME"             # S3 Tables namespace
TABLE_NAME = "YOUR_TABLE_NAME"

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

OUT_TABLE = f"{CATALOG_NAME}.{DB_NAME}.{TABLE_NAME}"

# ----------------------------------------------------------------------
# Create Sedona / Spark session and select the namespace
# ----------------------------------------------------------------------
# Your S3 Tables 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 catalog.
# The table bucket manages storage, so there is no warehouse path to set.

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()
        .withColumn("_source_file", F.lit(SOURCE_FILE))
        .withColumn("_ingested_at", F.current_timestamp())
)

# --------------------------------------------
# Create Iceberg table, then insert data
# --------------------------------------------
# S3 Tables manages storage for the table bucket, so no LOCATION is set.

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 clean_df.schema.fields
)

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

clean_df.createOrReplaceTempView("clean_source_data")

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

sedona.sql(f"""
CREATE TABLE {OUT_TABLE} (
{columns_sql}
)
USING iceberg
""")

sedona.sql(f"""
INSERT INTO {OUT_TABLE}
SELECT
{select_sql}
FROM clean_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)