aws--agent-toolkit-for-aws/plugins/aws-data-analytics/skills/ingesting-into-data-lake/references/schema-evolution.md

Schema Evolution and Nested Structure Handling Reference

This document describes expected approaches for handling schema evolution and nested JSON/struct data during imports.

Schema Evolution

What is Schema Evolution?

Schema evolution occurs when source data has columns that don't exist in the target table. This is common when:

• Source data schema changes over time (new fields added)
• Importing from multiple sources with different schemas
• Business requirements evolve and new data points are captured

Types of Schema Changes

Change Type	Example	Handling
New columns	Source has phone_number, table doesn't	ALTER TABLE ADD COLUMNS
Missing columns	Table has country, source doesn't	Use NULL or default value
Type changes	Source price is STRING, was INT	Type conflict resolution (see type-transformations.md)
Column rename	Source has customer_name, table has name	Manual mapping or user decision

Schema Evolution Workflow

1. Detect Schema Differences

# 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())

Expected output to user:

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.

2. Add New Columns via ALTER TABLE

With AWS CLI:

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

3. Handle Missing Columns

If source is missing columns that exist in the target table, two approaches:

Option 1: Use NULL for missing columns (recommended) — New rows will have NULL in these columns. Existing rows keep their values.

Option 2: Fail the import — Ensures data completeness. Requires source to have all columns.

Nested JSON Handling

Flatten vs Preserve Decision

When source data has nested structures:

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

Flattening Implementation

PySpark - Flatten Struct:

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")
)

PySpark - Explode Array:

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")
)

Athena SQL - Flatten with UNNEST:

-- 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);

Preserving Nested Structures

S3 Tables DDL with Nested Types:

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

Querying Nested Data:

-- 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);

PySpark - Write with Nested Types:

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

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

Array Handling Options

Implementation examples for each array handling approach:

Option 1: Keep as Array

Store as ARRAY<STRUCT<...>> in S3 Table. Query with UNNEST when needed. Preserves one-to-many relationships efficiently.

Option 2: Explode to Separate Rows

Each array element becomes its own row. Simple flat table structure. May create many duplicate rows if arrays are large.

Option 3: Create Separate Related Table

Store items in separate table (e.g., order_items). Link via foreign key. Normalized database design.

Complete Examples

Example 1: Schema Evolution

Before (existing table):

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

New Source Data adds columns: phone STRING, address STRING

After Evolution:

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

Result:

• Existing rows: customer_id=1, name="Alice", email="alice@example.com", phone=NULL, address=NULL
• New rows: customer_id=2, name="Bob", email="bob@example.com", phone="555-1234", address="123 Main St"

Example 2: Nested JSON with Flattening

Source JSON:

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

Flattened Table:

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

PySpark Code:

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")
)

Example 3: Nested JSON Preserved

Same Source, but preserved as nested:

Table Schema:

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

Query Example:

-- 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';

Evaluation Criteria

Schema Evolution

Detection:

• Compares source schema to existing table schema
• Identifies new, missing, and changed columns
• Reports differences clearly to user

Automatic Handling:

• New columns: Automatically executes ALTER TABLE ADD COLUMNS
• Missing columns: Uses NULL or asks user
• Type changes: Routes to type conflict resolution

Execution:

• ALTER TABLE commands are syntactically correct
• Uses appropriate Iceberg/S3 Tables syntax
• Verifies changes applied successfully

Nested JSON

Detection:

• Identifies STRUCT and ARRAY types in source
• Determines nesting depth
• Lists all nested fields clearly

User Choice:

• Presents flatten vs preserve options
• Explains pros/cons of each approach
• Waits for user decision

Implementation:

• Flatten: Provides complete PySpark/SQL with explode for arrays
• Preserve: Creates correct DDL with nested types
• Validates nested schema is correct

Query Examples:

• Shows how to query nested data
• Demonstrates struct field access (e.g., customer.name)
• Shows UNNEST/explode for arrays

Common Mistakes to Avoid

Recreating entire table when only ALTER TABLE ADD COLUMNS is needed Silently using NULL for missing columns without informing user Not asking user how to handle nested structures (flatten vs preserve) Incomplete flattening code (missing some nested fields) Incorrect DDL for nested types (wrong syntax) Not validating that ALTER TABLE succeeded Exploding arrays without explaining it creates multiple rows Not providing query examples for nested data access