Schema Evolution

# Get current table schema from Glue Catalog
import boto3
glue = boto3.client('glue')

response = glue.get_table(
    DatabaseName='my_database',
    Name='my_table'
)

existing_columns = {col['Name']: col['Type'] for col in response['Table']['StorageDescriptor']['Columns']}

# Compare with source schema
source_columns = {'customer_id': 'int', 'name': 'string', 'email': 'string', 'phone': 'string'}  # Inferred

new_columns = set(source_columns.keys()) - set(existing_columns.keys())
missing_columns = set(existing_columns.keys()) - set(source_columns.keys())

Schema Comparison:

Existing table columns: customer_id, name, email
Source data columns: customer_id, name, email, phone

New columns in source (will be added): phone
Missing columns in source (will be NULL): None

Schema evolution will automatically add new columns to the table.

aws athena start-query-execution \
  --query-string "ALTER TABLE \"catalog\".\"namespace\".\"table\" ADD COLUMNS (phone STRING)" \
  --query-execution-context Database=namespace \
  --result-configuration OutputLocation=s3://bucket/results/ \
  --region us-east-1

{
  "order_id": 12345,
  "customer": {
    "customer_id": 789,
    "name": "John Doe",
    "email": "john@example.com"
  },
  "items": [
    {"product_id": 456, "quantity": 2, "price": 29.99}
  ]
}

from pyspark.sql.functions import col

flattened_df = source_df.select(
    col("order_id"),
    col("customer.customer_id").alias("customer_id"),
    col("customer.name").alias("customer_name"),
    col("customer.email").alias("customer_email"),
    col("order_date"),
    col("total")
)

from pyspark.sql.functions import explode, col

# One row per item
exploded_df = source_df.select(
    col("order_id"),
    col("customer.customer_id").alias("customer_id"),
    explode(col("items")).alias("item")
).select(
    "order_id",
    "customer_id",
    col("item.product_id"),
    col("item.quantity"),
    col("item.price")
)

-- Create external table with nested types
CREATE EXTERNAL TABLE orders_nested (
  order_id BIGINT,
  customer STRUCT<customer_id: BIGINT, name: STRING, email: STRING>,
  items ARRAY<STRUCT<product_id: BIGINT, quantity: INT, price: DECIMAL(10,2)>>,
  order_date DATE,
  total DECIMAL(10,2)
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://bucket/orders/';

-- Flatten and insert
INSERT INTO "catalog"."namespace"."orders_flat"
SELECT
  order_id,
  customer.customer_id,
  customer.name AS customer_name,
  customer.email AS customer_email,
  item.product_id,
  item.quantity,
  item.price,
  order_date
FROM orders_nested
CROSS JOIN UNNEST(items) AS t(item);

CREATE TABLE "catalog"."namespace"."orders_nested" (
  order_id BIGINT,
  customer STRUCT<
    customer_id: BIGINT,
    name: STRING,
    email: STRING
  >,
  items ARRAY<STRUCT<
    product_id: BIGINT,
    quantity: INT,
    price: DECIMAL(10,2)
  >>,
  order_date DATE,
  total DECIMAL(10,2)
)
USING ICEBERG

-- Access struct fields
SELECT
  order_id,
  customer.name,
  customer.email,
  order_date
FROM "catalog"."namespace"."orders_nested"
WHERE customer.customer_id = 789;

-- Explode array in queries
SELECT
  order_id,
  item.product_id,
  item.quantity,
  item.price
FROM "catalog"."namespace"."orders_nested"
CROSS JOIN UNNEST(items) AS t(item);

# Preserve nested structure
source_df.writeTo(args['target_table']).append()

# No flattening needed - PySpark DataFrame schema maps directly to Iceberg

CREATE TABLE customers (
  customer_id INT,
  name STRING,
  email STRING
)

ALTER TABLE customers ADD COLUMNS (
  phone STRING,
  address STRING
);

{
  "user_id": 100,
  "profile": {
    "age": 30,
    "city": "Seattle"
  },
  "purchases": [
    {"item": "book", "amount": 20},
    {"item": "laptop", "amount": 1200}
  ]
}

user_id | age | city    | item   | amount
--------|-----|---------|--------|-------
100     | 30  | Seattle | book   | 20
100     | 30  | Seattle | laptop | 1200

from pyspark.sql.functions import explode, col

df = spark.read.json("s3://bucket/data.json")

flattened = df.select(
    col("user_id"),
    col("profile.age"),
    col("profile.city"),
    explode(col("purchases")).alias("purchase")
).select(
    "user_id",
    "age",
    "city",
    col("purchase.item"),
    col("purchase.amount")
)

CREATE TABLE user_purchases (
  user_id BIGINT,
  profile STRUCT<age: INT, city: STRING>,
  purchases ARRAY<STRUCT<item: STRING, amount: DECIMAL(10,2)>>
)

-- Get users from Seattle who bought laptops
SELECT
  user_id,
  profile.age,
  purchase.item,
  purchase.amount
FROM user_purchases
CROSS JOIN UNNEST(purchases) AS t(purchase)
WHERE profile.city = 'Seattle'
  AND purchase.item = 'laptop';