Building Modern Data Lakes with Amazon S3 Tables and Glue: A Complete ETL Implementation Guide with MCP Integration

SELECT * FROM "poc_main_dev" limit 10;

CREATE TABLE IF NOT EXISTS raw.productdb.products (
    product_id STRING,           -- Primary key for product identification
    product_name STRING,         -- Human-readable product name
    product_description STRING,  -- Detailed product information
    product_category STRING,     -- Product categorization
    product_price DOUBLE,        -- Current product price
    sale_date DATE,             -- Date of sale transaction
    sales_amount DOUBLE,        -- Total sales amount
    created_at TIMESTAMP,       -- Record creation timestamp
    updated_at TIMESTAMP        -- Last modification timestamp
) USING ICEBERG


ALTER TABLE raw.productdb.products
ADD PARTITION FIELD bucket(4, product_id)

def full_load(df):
    """Perform full load - overwrite all partitions"""
    print("Performing full load (overwrite partitions)...")

    # Use Iceberg's overwritePartitions for efficient full load
    df.writeTo(f"raw.{FULL_TABLE_NAME}").overwritePartitions()

    # Verify load
    final_count = spark.sql(f"SELECT COUNT(*) as count FROM raw.{FULL_TABLE_NAME}").collect()[0]['count']
    print(f"✓ Full load completed: {final_count:,} records in table")

def incremental_load(df):
    """Perform incremental load using MERGE operations"""
    merge_sql = f"""
    MERGE INTO raw.{FULL_TABLE_NAME} AS target
    USING source_data AS source
    ON target.product_id = source.product_id
    WHEN MATCHED THEN
      UPDATE SET
        product_name = source.product_name,
        product_description = source.product_description,
        product_category = source.product_category,
        product_price = source.product_price,
        sale_date = source.sale_date,
        sales_amount = source.sales_amount,
        updated_at = source.updated_at
    WHEN NOT MATCHED THEN
      INSERT (product_id, product_name, product_description,
              product_category, product_price, sale_date,
              sales_amount, created_at, updated_at)
      VALUES (source.product_id, source.product_name, source.product_description,
              source.product_category, source.product_price, source.sale_date,
              source.sales_amount, source.created_at, source.updated_at)
        spark.sql(merge_sql)