Ingesting Into Data Lake

Ingest into Data Lake

Move data from a source into a queryable table in the data lake. This skill assumes the source connection (if one is needed) already exists. For Glue connection setup or troubleshooting, delegate to connecting-to-data-source.

Philosophy

Default to S3 Tables unless the environment says otherwise. S3 Tables is the recommended target for new data lake work. If the user's catalog inventory shows they haven't adopted S3 Tables, recommend standard Iceberg on their existing general-purpose bucket instead of forcing them to change posture.

Common Tasks

You MUST execute commands using AWS MCP server tools when connected -- they provide validation, sandboxed execution, and audit logging. Fall back to AWS CLI only if MCP is unavailable. You MUST explain each step before executing.

Workflow

1. Verify Dependencies and Context

• You MUST check whether AWS MCP tools or AWS CLI are available and inform the user if missing
• You MUST confirm target AWS region and verify credentials with aws sts get-caller-identity
• For SageMaker Unified Studio project roles, note that target tables and connections may be scoped to the project. See the caller ARN detection pattern in querying-data-lake.

2. Classify the Source

If the user names Salesforce, ServiceNow, SAP, MongoDB, Kafka, or another SaaS/streaming source, decline -- these are not supported in this release.

If the source table is referenced by a fuzzy or business name ("migrate our orders table", "pull from the sales warehouse"), delegate to finding-data-lake-assets to resolve before proceeding.

3. Confirm Connection Exists (if applicable)

For JDBC, Snowflake, and BigQuery sources, a Glue connection is required. Check:

aws glue get-connection --name <CONNECTION_NAME> --region <REGION>

If the connection does not exist, stop and delegate to connecting-to-data-source to create and test it. Do not proceed with ingest until the connection is verified.

Local files, S3 files, DynamoDB, and catalog migration do not need a Glue connection.

4. Clarify the Target

You MUST ask the user (or suggest based on catalog inventory) before creating or writing to any table:

• Database/namespace: Does a specific target database exist? Or should one be created?
• Table: Existing table (append/merge) or new table (delegate to creating-data-lake-table)?
• Format: S3 Tables (default), standard Iceberg, or raw Parquet?

Inventory-aware defaults:

If you have already run exploring-data-catalog or can quickly check, use what exists:

• Account has an s3tablescatalog federated catalog and active table buckets: recommend S3 Tables
• Account has general-purpose buckets with Iceberg tables and no S3 Tables usage: recommend standard Iceberg on their existing bucket
• Account uses Parquet/ORC on S3 without Iceberg metadata: ask whether to adopt Iceberg now (recommend yes) or continue with raw files

Do not force S3 Tables on customers who haven't adopted it. See iceberg-catalog-config-and-usage.md.

Delegations from this step:

• Target table doesn't exist -> creating-data-lake-table
• Target database named by fuzzy term -> finding-data-lake-assets
• User doesn't know what exists -> exploring-data-catalog

5. Execute Source Workflow

Read the source-specific reference and follow its phases. Each is self-contained with job templates, gotchas, and troubleshooting:

• Local / S3 / JDBC / Snowflake / BigQuery / DynamoDB / catalog migration -- one reference per source

Common Glue 5.1 or higher job configuration and PySpark templates are shared in glue-job-config.md and glue-job-scripts.md.

6. Validate

Run all three, do not skip:

1. Row count matches expected (source vs target) 2. Null check on critical columns 3. Spot-check 3-5 sample rows

See data-quality-validation.md.

7. Schedule (if recurring)

For recurring pipelines, create a Glue Trigger with a cron schedule. See testing-and-scheduling.md. Simple single-step pipelines use Glue Triggers; multi-step with branching uses MWAA.

Argument Routing

• S3 path only: Infer one-time load, start Step 2 with S3 files
• Connection name: Start Step 3 with the named connection
• Table name: Start Step 4, ask whether this is source or target
• --target flag: Pre-fill the target format in Step 4
• No args: Walk through interactively

Gotchas

• S3 Tables requires Glue 5.1 or higher and --datalake-formats iceberg job argument
• All spark.sql.catalog.* config MUST go in --conf job arguments, never in spark.conf.set(). Glue 5.x throws AnalysisException: Cannot modify the value of a static config otherwise. See iceberg-catalog-config-and-usage.md for correct catalog configs.
• The warehouse parameter is required in S3 Tables catalog config. Without it Spark fails with "Cannot derive default warehouse location".
• Table and column names in S3 Tables MUST be all lowercase
• overwritePartitions() only replaces partitions present in the DataFrame -- for full refresh with deletes, use createOrReplace()
• Standard Iceberg targets MUST include a LOCATION clause; S3 Tables MUST NOT
• DynamoDB does not need a Glue connection -- do not attempt to create one
• Connection failures during ingest delegate back to connecting-to-data-source; do not debug network/credentials in this skill
• For target tables in SageMaker Unified Studio projects, ensure the project role has write access to the target namespace before the Glue job runs

Troubleshooting

See error-handling.md for the full catalog.

References

Source-specific

• local-upload.md -- Local files
• s3-files.md -- S3 files (CSV, JSON, Parquet, Avro, ORC)
• jdbc-ingest.md -- Oracle, SQL Server, PostgreSQL, MySQL, RDS, Aurora, Redshift
• snowflake-ingest.md -- Snowflake
• bigquery-ingest.md -- BigQuery
• dynamodb-ingest.md -- DynamoDB (export and Glue direct read)
• catalog-migration.md -- Existing Glue catalog tables (Hive, self-managed Iceberg)

Cross-cutting

• iceberg-catalog-config-and-usage.md -- S3 Tables, standard Iceberg, raw files: catalog config, engine access patterns
• glue-job-config.md -- Job sizing, monitoring, retry
• glue-job-scripts.md -- PySpark templates (append, upsert, custom SQL, full refresh)
• incremental-loading.md -- Watermark strategies
• testing-and-scheduling.md -- Glue Triggers, MWAA
• data-quality-validation.md -- Row counts, null checks, Glue Data Quality
• schema-evolution.md -- ALTER TABLE ADD COLUMNS, nested JSON
• type-transformations.md -- Type conflict resolution
• format-specific-loading.md -- CSV/JSON/Parquet/Avro/ORC specifics
• athena-loading.md -- Athena INSERT INTO as simple-load fallback
• error-handling.md -- Ingest errors (connection errors delegate to connecting-to-data-source)
• upload-options.md -- aws s3 cp vs sync, multipart

Migration-specific

• ctas-patterns.md -- Athena CTAS syntax and partition transforms
• glue-etl-migration.md -- Large-table migration via Glue 5.1 or higher PySpark
• migration-validation.md -- Full validation checklist
• migration-troubleshooting.md -- CTAS failures, visibility, partitions

JDBC-specific

• jdbc-schema-discovery.md -- Crawler, direct inspection, custom SQL
• jdbc-performance.md -- Parallel reads, partitioning

Data Loading via Athena INSERT INTO

Fallback approach for simple one-time data loads when Glue ETL is unavailable or unnecessary.

Step 1: Create External Table for Source

Create a temporary external table pointing to source files in S3.

CSV

CREATE EXTERNAL TABLE temp_source_<timestamp> (
  customer_id INT,
  first_name STRING,
  last_name STRING,
  email STRING,
  signup_date STRING
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS TEXTFILE
LOCATION 's3://<bucket>/<prefix>/'
TBLPROPERTIES ('skip.header.line.count'='1');

JSON

CREATE EXTERNAL TABLE temp_source_<timestamp> (
  order_id BIGINT,
  customer_id BIGINT,
  order_date STRING,
  total DECIMAL(10,2)
)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
LOCATION 's3://<bucket>/<prefix>/';

Parquet / ORC

CREATE EXTERNAL TABLE temp_source_<timestamp> (
  event_id BIGINT,
  event_type STRING,
  timestamp TIMESTAMP
)
STORED AS PARQUET  -- or ORC
LOCATION 's3://<bucket>/<prefix>/';

Step 2: Transform and Insert

INSERT INTO "<catalog>"."<namespace>"."<target_table>"
SELECT
  CAST(customer_id AS BIGINT) AS customer_id,
  first_name,
  last_name,
  email,
  DATE_PARSE(signup_date, '%Y-%m-%d') AS signup_date
FROM temp_source_<timestamp>
WHERE customer_id IS NOT NULL

For detailed type casting, date parsing, null handling, and boolean conversion patterns, see type-transformations.md.

Execute via CLI

QUERY_ID=$(aws athena start-query-execution \
  --query-string "<INSERT INTO query>" \
  --query-execution-context Database=<namespace> \
  --result-configuration OutputLocation=s3://<results-bucket>/ \
  --region <region> \
  --query 'QueryExecutionId' --output text)

aws athena get-query-execution --query-execution-id "$QUERY_ID" --region <region>

Step 3: Validate

-- Row count
SELECT COUNT(*) as row_count FROM "<catalog>"."<namespace>"."<target_table>";

-- Spot check
SELECT * FROM "<catalog>"."<namespace>"."<target_table>" LIMIT 10;

-- Null check on critical columns
SELECT
  SUM(CASE WHEN customer_id IS NULL THEN 1 ELSE 0 END) as null_ids,
  COUNT(*) as total
FROM "<catalog>"."<namespace>"."<target_table>";

Step 4: Clean Up

DROP TABLE IF EXISTS temp_source_<timestamp>;

Large Datasets

If Athena times out (30-minute limit):

1. Batch by partition: Load one month/day at a time 2. Switch to Glue ETL: Better for datasets > 1GB — handles larger data with more workers, provides monitoring and retries

Limitations

BigQuery Ingest

Move data from Google BigQuery into the data lake. Assumes a Glue BIGQUERY connection exists. If not, delegate to connecting-to-data-source.

Contents

• Prerequisites
• Read Pattern
• Incremental Loading
• Partition Decorators
• Type Mapping
• Further Reading

Prerequisites

• Glue connection of type BIGQUERY with service account credentials in Secrets Manager
• GCP project ID and source table (full form: project.dataset.table)
• Target table in the data lake
• Egress from the Glue subnet to bigquery.googleapis.com (public internet or Google Private Service Connect)

Read Pattern

bigquery_df = glueContext.create_dynamic_frame.from_options(
    connection_type="bigquery",
    connection_options={
        "connectionName": args['connection_name'],
        "parentProject": args['gcp_project'],
        "sourceType": "table",
        "table": "my_dataset.customers"
    }
).toDF()

For custom SQL:

connection_options={
    "connectionName": args['connection_name'],
    "parentProject": args['gcp_project'],
    "sourceType": "query",
    "query": "SELECT id, name, updated_at FROM `project.dataset.customers` WHERE country = 'US'"
}

BigQuery billing note: the query reads bytes from table storage. Filter aggressively at source to minimize bytes scanned.

Incremental Loading

BigQuery has strong timestamp semantics. Watermark columns commonly used:

• Application-maintained updated_at / last_modified
• BigQuery-maintained _PARTITIONTIME / _PARTITIONDATE on partitioned tables
• INFORMATION_SCHEMA.PARTITIONS.last_modified_time for partition-level freshness

Example incremental read with watermark filter:

query = f"""
SELECT *
FROM `{project}.{dataset}.{table}`
WHERE updated_at > TIMESTAMP('{last_watermark}')
"""

See incremental-loading.md for watermark storage.

Partition Decorators

For time-partitioned BigQuery tables, use partition decorators to target specific partitions and reduce bytes scanned:

# Read only 2026-04 partitions
query = f"""
SELECT *
FROM `{project}.{dataset}.{table}`
WHERE _PARTITIONTIME BETWEEN TIMESTAMP('2026-04-01') AND TIMESTAMP('2026-04-30')
"""

Clustered tables benefit similarly from filter push-down on clustering columns. Check clustering:

SELECT clustering_fields FROM `<project>.<dataset>.INFORMATION_SCHEMA.TABLES` WHERE table_name = '<table>';

Type Mapping

BIGNUMERIC (up to 76.38 precision) exceeds Iceberg DECIMAL's 38-digit cap. For full-precision needs, store as STRING and cast on read.

Further Reading

• AWS Glue: Creating a BigQuery connection
• AWS Glue: Creating a BigQuery source node
• BigQuery partitioned tables
• BigQuery clustered tables

Catalog Migration to S3 Tables

Migrate existing Glue Data Catalog tables into Amazon S3 Tables. Source tables can be Hive-format, self-managed Iceberg, or any format Athena can read. The result is a fully managed S3 Table with automatic compaction, snapshot management, and multi-engine access.

Reference Documentation

• ctas-patterns.md -- Athena CTAS syntax for S3 Tables, format options, partition transforms
• migration-validation.md -- Row count, schema, and data integrity checks
• glue-etl-migration.md -- Glue 5.1 or higher PySpark migration for large tables
• migration-troubleshooting.md -- Common errors and fixes

Why Migrate?

Self-managed Iceberg and Hive tables require manual compaction, snapshot cleanup, and storage optimization. S3 Tables handles all of this automatically. Migration also enables the four-part catalog hierarchy (s3tablescatalog/<bucket>/<namespace>/<table>) for unified access from Athena, EMR, Redshift, and Spark.

Note: The target for catalog migration is always S3 Tables -- that is the purpose of this workflow.

Workflow

Phase 1: Understand the Source

1. Identify the source table: Get the fully qualified name (database.table or catalog.database.table). If the user gives a fuzzy or business name ("our orders table", "the sales data"), delegate to the finding-data-lake-assets skill to resolve it before continuing -- the rest of this workflow assumes a concrete reference. 2. Inspect the source:

• With MCP: Use aws-mcp to get table metadata (format, location, schema, partitions)
• Without MCP: aws glue get-table --database-name <db> --name <table>

3. Classify the source format:

• Hive (CSV, Parquet, ORC, JSON, Avro): Standard external table backed by S3 general purpose bucket
• Self-managed Iceberg: Iceberg table in general purpose bucket with manual maintenance
• Other: Any format Athena can query (federated sources, etc.)

4. Assess size and complexity:

• Small/medium (under ~100 GB, simple schema): Path A (Athena CTAS) -- single SQL statement
• Large (over ~100 GB, complex transforms, or needs scheduling): Path B (Glue ETL)
• Partitioned source: Note partition columns and strategy for conversion

Phase 2: Prepare the Target

1. Ensure table bucket exists: Check with aws s3tables list-table-buckets. If none, delegate to creating-data-lake-table Phase 2. 2. Ensure analytics integration is enabled: Verify s3tablescatalog exists. Delegate to creating-data-lake-table Phase 2, step 4 if not set up. 3. Create or select namespace: Use existing or create new via aws s3tables create-namespace. 4. Plan partition strategy: Iceberg supports hidden partition transforms (day(), month(), year(), hour(), bucket()). Recommend converting Hive-style explicit partition columns to Iceberg transforms where possible.

Phase 3: Migrate the Data

Path A: Athena CTAS (default for small/medium tables)

Single SQL statement that creates the S3 Table and populates it in one step. See ctas-patterns.md for full syntax and examples.

Key points:

• Target path: "s3tablescatalog/<table_bucket_name>"."<namespace>"."<new_table_name>"
• Default format: PARQUET. Also supports AVRO, ORC.
• Use Iceberg partition transforms (day(), month(), bucket()) instead of Hive-style explicit partition columns.
• No LOCATION clause -- S3 Tables manages storage.
• Table and column names must be all lowercase.
• Source catalog for default GDC tables is awsdatacatalog.
• Add WHERE filters to migrate subsets or batch large migrations.

Path B: Glue ETL (for large tables or complex transforms)

Use when CTAS would time out, when transforms are complex, or when the migration needs to be scheduled/repeatable.

1. Create PySpark script that reads from source and writes to S3 Table 2. Create Glue 5.1 or higher job with --datalake-formats iceberg and --conf catalog config 3. Run and monitor the job

See glue-etl-migration.md for job configuration, PySpark script template, and catalog setup.

Phase 4: Validate the Migration

Run all of these checks -- do not skip any:

1. Row count comparison:

   SELECT 'source' AS tbl, COUNT(*) AS cnt FROM "<source_catalog>"."<source_db>"."<source_table>"
   UNION ALL
   SELECT 'target' AS tbl, COUNT(*) AS cnt FROM "s3tablescatalog/<bucket>"."<namespace>"."<new_table>"

2. Schema comparison: Verify column names, types, and order match expectations. Minor type promotions (e.g., int to bigint) are acceptable.

3. Spot-check data: Compare a sample of rows between source and target, focusing on:

• Boundary values (min/max of numeric and date columns)
• Null counts per column
• Distinct counts on key columns

4. Partition verification (if partitioned):

   SELECT <partition_column>, COUNT(*) FROM "s3tablescatalog/<bucket>"."<namespace>"."<new_table>"
   GROUP BY 1 ORDER BY 1

See migration-validation.md for the full checklist.

Phase 5: Post-Migration Guidance

After validation passes:

1. Update downstream consumers: Provide the new table path for queries, dashboards, and ETL jobs. 2. Recommend keeping the source table temporarily as a rollback option. Suggest a retention period (e.g., 30 days). 3. Do NOT drop the source table. Warn the user and let them decide when to clean up. 4. Evaluate table lineage: If the source table has lineage present, use it to recommend next-steps for producers and consumers.

Gotchas

• Athena CTAS has a 100-partition limit per statement. For sources with more than 100 partitions, either migrate in batches with WHERE filters or use Glue ETL (Path B).
• CTAS creates a new table -- it does not do an in-place conversion. The source table remains unchanged.
• Column names with uppercase letters will cause the target table to be invisible to analytics services. Always lowercase column names in the SELECT: SELECT upper_Col AS upper_col.
• Self-managed Iceberg tables may have schema evolution history (added/renamed columns). CTAS captures the current schema only -- historical evolution is not preserved.
• Hive tables with complex SerDe configurations (custom delimiters, regex SerDe) should be tested with a small CTAS first to verify Athena can read them correctly. Glue will often read things Athena cannot. Try Glue if Athena fails.
• Time travel on the source Iceberg table is lost after migration. The S3 Table starts fresh with its own snapshot history.

Troubleshooting

See migration-troubleshooting.md for common errors and fixes covering CTAS failures, validation mismatches, visibility issues, and partition problems.

Athena CTAS Patterns for S3 Tables Migration

Basic Migration (no partitions)

CREATE TABLE "s3tablescatalog/my-bucket"."my_namespace"."customers"
WITH (format = 'PARQUET') AS
SELECT * FROM "awsdatacatalog"."legacy_db"."customers"

Migration with Iceberg Partition Transforms

Convert Hive-style explicit partitions to Iceberg hidden partitions:

-- Source has explicit year/month/day columns from Hive partitioning
-- Target uses Iceberg day() transform on the timestamp column
CREATE TABLE "s3tablescatalog/my-bucket"."analytics"."events"
WITH (
    format = 'PARQUET',
    partitioning = ARRAY['day(event_timestamp)']
) AS
SELECT
    event_id,
    user_id,
    event_type,
    event_timestamp,
    payload
FROM "awsdatacatalog"."raw_db"."events_hive"

Available Partition Transforms

Batched Migration (over 100 partitions)

Athena CTAS has a 100-partition limit per statement. Migrate in batches:

-- Batch 1: 2023 data
CREATE TABLE "s3tablescatalog/my-bucket"."ns"."orders"
WITH (format = 'PARQUET', partitioning = ARRAY['month(order_date)']) AS
SELECT * FROM "awsdatacatalog"."sales"."orders"
WHERE order_date >= DATE '2023-01-01' AND order_date < DATE '2024-01-01'

-- Batch 2+: INSERT INTO for subsequent years
INSERT INTO "s3tablescatalog/my-bucket"."ns"."orders"
SELECT * FROM "awsdatacatalog"."sales"."orders"
WHERE order_date >= DATE '2024-01-01' AND order_date < DATE '2025-01-01'

Migration with Column Transformations

CREATE TABLE "s3tablescatalog/my-bucket"."clean"."users"
WITH (format = 'PARQUET') AS
SELECT
    user_id,
    LOWER(email) AS email,
    COALESCE(display_name, username) AS name,
    CAST(created_at AS timestamp) AS created_at,
    CASE WHEN status = 'A' THEN 'active' ELSE 'inactive' END AS status
FROM "awsdatacatalog"."legacy"."users_raw"

Cross-Catalog Migration (self-managed Iceberg)

CREATE TABLE "s3tablescatalog/my-bucket"."analytics"."transactions"
WITH (
    format = 'PARQUET',
    partitioning = ARRAY['day(transaction_date)']
) AS
SELECT * FROM "awsdatacatalog"."iceberg_db"."transactions_selfmanaged"

Format Options

Data Quality and Validation

Complete guide for validating data quality during and after import into S3 Tables.

Overview

Data quality validation ensures that loaded data meets expected standards for completeness, accuracy, and consistency. This reference covers:

• Glue Data Quality rules integration
• Basic post-load validation queries
• Common validation patterns
• Troubleshooting quality issues

Glue Data Quality Rules

Integrate Glue Data Quality rules directly into your ETL jobs for automated validation during the load.

Basic Integration

Add to your Glue job PySpark script:

from awsglue.data_quality import DataQualityEvaluationOptions, DataQualityEvaluator
from awsglue.dynamicframe import DynamicFrame

# Define data quality rules
rules = """
    Rules = [
        RowCount > 0,
        ColumnCount == <expected_count>,
        ColumnValues "<column_name>" Completeness > 0.95,
        ColumnValues "<numeric_column>" between <min> and <max>,
        IsPrimaryKey "<id_column>",
        Uniqueness "<id_column>" > 0.99
    ]
"""

# Evaluate data quality
evaluator = DataQualityEvaluator(
    glueContext,
    rules,
    DynamicFrame.fromDF(transformed_df, glueContext, "check")
)
result = evaluator.evaluate()

# Fail job if quality checks don't pass
if result.overallResult != "PASS":
    raise Exception(f"Data quality check failed: {result}")

Available Data Quality Rules

Complete Example with Multiple Rules

from awsglue.data_quality import DataQualityEvaluationOptions, DataQualityEvaluator
from awsglue.dynamicframe import DynamicFrame

# Define comprehensive data quality rules
rules = """
    Rules = [
        # Basic structure checks
        RowCount > 100,
        ColumnCount == 8,

        # Completeness checks
        IsComplete "customer_id",
        IsComplete "order_date",
        ColumnValues "email" Completeness > 0.90,

        # Uniqueness checks
        IsPrimaryKey "order_id",
        Uniqueness "customer_id" > 0.80,

        # Value range checks
        ColumnValues "quantity" between 1 and 1000,
        ColumnValues "price" between 0.01 and 10000.00,
        ColumnValues "order_date" >= "2023-01-01",

        # Statistical checks
        Mean "price" between 10.0 and 500.0,
        StandardDeviation "quantity" < 100.0,

        # Categorical checks
        ColumnValues "status" in ["pending", "completed", "cancelled"],
        DistinctValuesCount "status" == 3
    ]
"""

# Convert DataFrame to DynamicFrame for evaluation
dynamic_frame = DynamicFrame.fromDF(transformed_df, glueContext, "quality_check")

# Create evaluation options
eval_options = DataQualityEvaluationOptions(
    publishCloudWatchMetrics=True,
    publishResultsToCloudWatch=True
)

# Evaluate data quality
evaluator = DataQualityEvaluator(glueContext, rules, dynamic_frame, eval_options)
result = evaluator.evaluate()

# Check results
if result.overallResult != "PASS":
    # Log failed rules
    for rule_result in result.ruleResults:
        if rule_result.result == "FAIL":
            print(f"Failed rule: {rule_result.rule}")
            print(f"Failure reason: {rule_result.failureReason}")

    # Fail the job
    raise Exception(f"Data quality check failed: {result.overallResult}")
else:
    print("All data quality checks passed!")

Conditional Quality Checks

Only fail on critical issues:

# Define critical vs warning rules
critical_rules = """
    Rules = [
        IsPrimaryKey "order_id",
        IsComplete "customer_id",
        RowCount > 0
    ]
"""

warning_rules = """
    Rules = [
        ColumnValues "email" Completeness > 0.90,
        Mean "price" between 10.0 and 500.0
    ]
"""

# Evaluate critical rules (fail on failure)
critical_result = DataQualityEvaluator(glueContext, critical_rules, dynamic_frame).evaluate()
if critical_result.overallResult != "PASS":
    raise Exception(f"Critical data quality check failed")

# Evaluate warning rules (log but don't fail)
warning_result = DataQualityEvaluator(glueContext, warning_rules, dynamic_frame).evaluate()
if warning_result.overallResult != "PASS":
    print(f"Warning: Non-critical data quality issues detected")
    for rule_result in warning_result.ruleResults:
        if rule_result.result == "FAIL":
            print(f"  - {rule_result.rule}: {rule_result.failureReason}")

Basic Validation Without Glue Data Quality

Even without Glue Data Quality, perform basic checks using Athena queries after the load.

1. Row Count Validation

Verify data was loaded:

-- Count rows in target table
SELECT COUNT(*) as row_count
FROM "<catalog>"."<namespace>"."<table>"

Compare with source row count (if available).

2. Null Checks

Verify critical columns aren't mostly null:

-- Check null percentages for critical columns
SELECT
    COUNT(*) as total_rows,
    SUM(CASE WHEN customer_id IS NULL THEN 1 ELSE 0 END) as null_customer_id,
    SUM(CASE WHEN order_date IS NULL THEN 1 ELSE 0 END) as null_order_date,
    SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END) as null_amount,
    -- Calculate percentages
    CAST(SUM(CASE WHEN customer_id IS NULL THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) * 100 as pct_null_customer_id,
    CAST(SUM(CASE WHEN order_date IS NULL THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) * 100 as pct_null_order_date,
    CAST(SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) * 100 as pct_null_amount
FROM "<catalog>"."<namespace>"."<table>"

3. Type Validation

Sample check that types converted correctly:

-- Sample data to verify types
SELECT *
FROM "<catalog>"."<namespace>"."<table>"
LIMIT 100

Look for:

• Dates that look like strings (e.g., "2024-01-15" instead of DATE)
• Numbers that are actually strings
• Truncated decimals
• Unexpected null values

4. Duplicate Detection

Check for unexpected duplicates on key columns:

-- Find duplicate order_ids
SELECT
    order_id,
    COUNT(*) as duplicate_count
FROM "<catalog>"."<namespace>"."<table>"
GROUP BY order_id
HAVING COUNT(*) > 1
ORDER BY duplicate_count DESC
LIMIT 100

5. Value Range Checks

Verify values are within expected ranges:

-- Check value ranges
SELECT
    MIN(order_date) as min_date,
    MAX(order_date) as max_date,
    MIN(amount) as min_amount,
    MAX(amount) as max_amount,
    MIN(quantity) as min_quantity,
    MAX(quantity) as max_quantity
FROM "<catalog>"."<namespace>"."<table>"

6. Categorical Value Checks

Verify categorical columns have expected values:

-- Check distinct values in status column
SELECT
    status,
    COUNT(*) as count
FROM "<catalog>"."<namespace>"."<table>"
GROUP BY status
ORDER BY count DESC

Expected values should match source data categories.

7. Statistical Checks

Get basic statistics:

-- Calculate basic statistics
SELECT
    COUNT(*) as total_rows,
    AVG(amount) as avg_amount,
    STDDEV(amount) as stddev_amount,
    APPROX_PERCENTILE(amount, 0.5) as median_amount,
    APPROX_PERCENTILE(amount, 0.95) as p95_amount
FROM "<catalog>"."<namespace>"."<table>"

Validation Reporting

Present Results to User

After running validation queries, present results clearly:

Data Load Validation Report:
✓ Row count: 1,234,567 rows loaded
✓ Null checks:
  - customer_id: 0% null (expected: 0%)
  - order_date: 0.1% null (acceptable)
  - amount: 2.3% null (within threshold)
✓ Duplicates: No duplicate order_ids found
✓ Value ranges:
  - order_date: 2023-01-01 to 2024-12-31 (expected)
  - amount: $0.01 to $9,999.99 (valid range)
  - quantity: 1 to 500 (valid range)
✓ Categorical values:
  - status: pending (45%), completed (50%), cancelled (5%)
⚠ Warning: email column has 10% null values (target: < 5%)

Overall: PASS with warnings

Handle Failures

When validation fails:

# In Glue job script
if result.overallResult != "PASS":
    failure_summary = []
    for rule_result in result.ruleResults:
        if rule_result.result == "FAIL":
            failure_summary.append(f"  - {rule_result.rule}: {rule_result.failureReason}")

    error_message = "Data quality validation failed:\n" + "\n".join(failure_summary)
    print(error_message)

    # Optionally send notification or write to error table
    # Then fail the job
    raise Exception(error_message)

Common Validation Patterns

Pre-Load Validation

Before loading, validate source data:

# Sample source data
sample_df = spark.read.format("csv").option("header", "true").load(source_path).limit(1000)

# Check structure
print(f"Row count: {sample_df.count()}")
print(f"Column count: {len(sample_df.columns)}")
print(f"Columns: {sample_df.columns}")
print(f"Schema: {sample_df.printSchema()}")

# Check for issues
null_counts = sample_df.select([
    (col(c).isNull().cast("int")).alias(c) for c in sample_df.columns
]).groupBy().sum()

print("Null counts in sample:")
null_counts.show()

Post-Load Reconciliation

Compare source and target row counts:

# Count source rows
source_count = spark.read.format("csv").option("header", "true").load(source_path).count()

# Count target rows
target_count = spark.sql(f"SELECT COUNT(*) FROM {target_table}").collect()[0][0]

# Verify match
if source_count != target_count:
    print(f"Row count mismatch: source={source_count}, target={target_count}")
    raise Exception("Row count mismatch detected")
else:
    print(f"Row count validation passed: {target_count} rows")

Troubleshooting Quality Issues

Issue: High Null Percentage

Symptoms: More nulls than expected in columns Possible causes:

• Source data quality issues
• Type conversion failures (strings that can't be parsed as numbers)
• Column mapping errors

Solutions:

1. Check source data for null values 2. Verify type conversions are correct 3. Add explicit null handling in transformation

Issue: Duplicate Keys

Symptoms: Primary key column has duplicates Possible causes:

• Source data has duplicates
• Multiple loads without deduplication
• Partition keys included in data

Solutions:

1. Add deduplication logic to Glue job 2. Use window functions to keep only latest record 3. Investigate source data quality

Issue: Value Range Violations

Symptoms: Values outside expected ranges Possible causes:

• Source data contains outliers
• Type conversion errors
• Unit mismatches (e.g., dollars vs cents)

Solutions:

1. Add filtering or capping in transformation 2. Verify unit conversions 3. Add validation rules to reject bad data

Best Practices

1. Start with basic checks: Row count and null checks catch most issues 2. Add rules incrementally: Begin with critical rules, expand over time 3. Use sampling for large datasets: Validate sample before full load 4. Publish metrics to CloudWatch: Enable monitoring and alerting 5. Document thresholds: Make quality expectations explicit 6. Handle warnings separately from errors: Not all issues should fail the job 7. Test quality rules: Ensure rules actually catch bad data

Summary

Data quality validation workflow:

1. Pre-load validation: Sample and inspect source data 2. In-load validation: Use Glue Data Quality rules during ETL 3. Post-load validation: Run Athena queries to verify results 4. Reconciliation: Compare source and target row counts 5. Reporting: Present clear validation results to user

With comprehensive validation, you can ensure data loaded into S3 Tables meets quality standards.

DynamoDB Ingest

Import DynamoDB tables into the data lake. DynamoDB is unique among sources: no Glue connection needed, schemaless items, and no natural watermark column.

Contents

• Method Selection
• Native Export (Path A)
• Glue Direct Read (Path B)
• Schema Flattening
• Incremental Strategies
• Throughput Guidance
• Gotchas

Method Selection

Assess the table:

aws dynamodb describe-table --table-name <TABLE>

Note item count, table size, billing mode, and PITR status.

Native Export (Path A)

Recommended for medium/large tables. Uses no read capacity.

Export Command

aws dynamodb export-table-to-point-in-time \
  --table-arn arn:aws:dynamodb:<REGION>:<ACCOUNT>:table/<TABLE> \
  --s3-bucket <EXPORT_BUCKET> \
  --s3-prefix exports/<TABLE>/ \
  --export-format DYNAMODB_JSON \
  --export-type FULL_EXPORT

Export formats:

• DYNAMODB_JSON (default) -- each item as JSON with type descriptors like {"S": "value"}
• ION -- Amazon Ion, more compact, handles binary natively

Monitoring

aws dynamodb describe-export --export-arn <EXPORT_ARN>

States: IN_PROGRESS, COMPLETED, FAILED. Large tables take minutes to hours.

Output Structure

s3://<bucket>/exports/<table>/AWSDynamoDB/<export-id>/
  manifest-summary.json
  manifest-files.json
  data/                    (gzipped JSON or Ion)

Read Export in Glue

export_df = spark.read.json("s3://<bucket>/exports/<table>/AWSDynamoDB/<export-id>/data/")
# Items are nested in type descriptors -- flatten per Schema Flattening below

Native export items are wrapped in DynamoDB type descriptors ({"S": "value"}, {"N": "123"}). Unwrap before flattening:

# Native export items are wrapped in type descriptors -- unwrap before flattening:
flat_df = export_df.select(
    col("Item.pk.S").alias("partition_key"),
    col("Item.name.S").alias("name"),
    col("Item.age.N").cast("bigint").alias("age")
)

Incremental Export

Requires PITR enabled on the source table.

aws dynamodb export-table-to-point-in-time \
  --table-arn <arn> \
  --s3-bucket <bucket> \
  --export-type INCREMENTAL_EXPORT \
  --incremental-export-specification '{"ExportFromTime":"<last>","ExportToTime":"<now>","ExportViewType":"NEW_AND_OLD_IMAGES"}'

Glue Direct Read (Path B)

For small tables. No connection needed -- Glue reads DynamoDB via AWS APIs with the Glue job role's permissions.

dynamodb_df = glueContext.create_dynamic_frame.from_options(
    connection_type="dynamodb",
    connection_options={
        "dynamodb.input.tableName": "<TABLE>",
        "dynamodb.throughput.read.percent": "0.5"
    }
).toDF()

# After flattening, write to target (see iceberg-catalog-config-and-usage.md for path syntax)
flat_df.writeTo("s3tablescatalog.<namespace>.<table>").append()

Options:

Schema Flattening

Applies to Glue direct-read (Path B) output. For native export (Path A) output, use the type-descriptor unwrapping pattern shown above.

DynamoDB type to Iceberg:

Strategy options

Top-level only (simplest):

flat_df = dynamodb_df.select(
    col("pk").alias("partition_key"),
    col("name").cast("string"),
    col("created_at").cast("timestamp")
)

Flatten one level:

flat_df = dynamodb_df.select(
    col("pk").alias("user_id"),
    col("profile.first_name").alias("first_name"),
    col("address.city").alias("city")
)

Preserve as STRUCT:

flat_df = dynamodb_df.select(col("pk"), col("profile"), col("tags"))

Serialize complex types to JSON:

from pyspark.sql.functions import to_json
flat_df = dynamodb_df.select(col("pk"), to_json(col("metadata")).alias("metadata_json"))

Sample items for schema inference

aws dynamodb scan --table-name <TABLE> --limit 10 --output json

Or in Spark:

sample = dynamodb_df.limit(100).toPandas()
all_columns = set()
for _, row in sample.iterrows():
    all_columns.update(row.dropna().index.tolist())

Missing attributes

from pyspark.sql.functions import coalesce, lit
flat_df = dynamodb_df.select(
    col("pk"),
    coalesce(col("email"), lit("")).alias("email"),
    coalesce(col("status"), lit("unknown")).alias("status")
)

Incremental Strategies

Scheduled full export: EventBridge rule triggers Lambda that runs export-table-to-point-in-time then a Glue job. Simple, captures deletes.

DynamoDB Streams: Enable with --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES. Lambda consumes stream, writes to S3 or target. 24-hour stream retention -- Lambda must keep up.

Application watermark: If items have last_modified attribute, filter in Glue: dynamodb_df.filter(f"last_modified > '{last_watermark}'"). Requires app cooperation and consumes read capacity.

Full refresh: For small tables, dynamodb_df.writeTo(target).using("iceberg").createOrReplace(). Do NOT use overwritePartitions() -- it only replaces partitions present in the DataFrame, leaving deleted items as stale data.

Throughput Guidance

Gotchas

• Native export consumes no read capacity -- always prefer for tables over 1 GB
• Glue direct reads with high read.percent can throttle production traffic
• DynamoDB Number is arbitrary precision -- decide BIGINT vs DECIMAL based on actual values
• Binary (B) attributes rarely useful in analytics -- exclude unless required
• DynamoDB Streams retention is 24 hours -- if the consumer falls behind, data is lost
• Incremental export requires PITR enabled
• overwritePartitions() does NOT delete partitions missing from the source DataFrame

Error Handling and Troubleshooting

Complete guide for handling common errors and issues during data import into S3 Tables.

Overview

This reference covers errors encountered during the data import workflow. Errors are organized by workflow phase and severity.

Connection errors are out of scope for this skill. JDBC/Snowflake/BigQuery connection failures (timeouts, auth failures, driver not found, SSL errors) belong to connecting-to-data-source. When a Glue job fails with a connection-level error, delegate to that skill's troubleshooting rather than debugging here.

Common Issues by Category

Schema Mismatch Errors

Symptoms:

• Type conversion failures during load
• Column count mismatches between source and target
• Data truncation warnings
• Null values where not expected

Root Causes:

• Source data types don't match target Iceberg types
• New columns in source not present in target table
• Missing columns in source that exist in target
• Incompatible type conversions (e.g., string → int with non-numeric values)

Solutions:

1. Type mismatch - can cast safely:

• Present conflict to user with example values
• Offer to add explicit CAST in transformation
• See type-transformations.md for casting patterns

2. Type mismatch - cannot cast:

• Show sample problematic values
• Options:
• Filter out invalid rows
• Store as STRING and convert later
• Fix source data and re-import
• Let user decide based on data importance

3. New columns in source:

• Suggest schema evolution via ALTER TABLE ADD COLUMNS
• Show proposed schema change
• Execute evolution if user approves
• See schema-evolution.md

4. Missing columns in source:

• Ask user how to handle:
• Default values (e.g., NULL, 0, empty string)
• Skip these columns (if nullable)
• Fail the load (if columns are critical)

Example Error Message to Present:

Schema Mismatch Detected:
- Column "age": Source type STRING, Target type INT
  Sample values: "25", "thirty", "42", "unknown"
  Issue: Values "thirty" and "unknown" cannot convert to INT

Options:
1. Filter out rows with non-numeric ages (loses ~5% of data)
2. Store age as STRING in target table (requires schema change)
3. Replace non-numeric values with NULL (preserves all rows)

Which approach would you prefer?

Permission Errors

Symptoms:

• Access Denied errors from AWS services
• IAM role assumption failures
• S3 bucket access errors
• Glue job fails with permission errors

Root Causes:

• Missing IAM policies on Glue service role
• S3 bucket policies blocking access
• S3 Tables permissions not configured
• Cross-account access issues

Solutions:

1. Glue service role missing policies:

• Check if role has AWSGlueServiceRole managed policy
• Check if role has S3 read/write permissions
• Check if role has S3 Tables inline policy
• See iam-role-management.md for complete setup

2. S3 bucket access denied:

• Verify IAM role has s3:GetObject, s3:ListBucket on source bucket
• Verify IAM role has s3:PutObject on script/results buckets
• Check S3 bucket policies don't block the role
• For cross-account: verify bucket policy allows role ARN

3. S3 Tables access denied:

• Verify inline policy includes:
• s3tables:PutTableData
• s3tables:GetTableMetadataLocation
• s3tables:GetTable
• s3tables:UpdateTableMetadataLocation
• Verify resource ARN matches table bucket structure
• See iam-role-management.md

4. Athena query execution errors:

• Verify workgroup has output location configured
• Verify IAM has athena:StartQueryExecution
• Verify IAM has s3:PutObject on results bucket

Example Error Message to Present:

Permission Error Detected:
Glue job failed with: "Access Denied" when writing to table

Root cause: IAM role "GlueServiceRole-import" is missing S3 Tables permissions

Required actions:
1. Add inline policy to role with s3tables:PutTableData permission
2. Resource ARN should be: arn:aws:s3tables:us-east-1:123456789012:bucket/my-table-bucket/namespace/my-namespace/table/*

Would you like me to add this policy to the role?

Data Quality Failures

Symptoms:

• Glue Data Quality rules fail
• Row counts don't match expected
• High null percentages in critical columns
• Duplicate primary keys detected

Root Causes:

• Source data quality issues
• Incorrect transformation logic
• Schema inference errors
• Data quality rules too strict

Solutions:

1. Row count mismatch:

• Compare source row count vs target row count
• Check Glue job logs for filtering or errors
• Verify no duplicate writes occurred
• Check if partitioned data was partially loaded

2. High null percentage:

• Show which columns have unexpected nulls
• Check if type conversion failures resulted in nulls
• Ask user if nulls are acceptable or if source needs fixing
• Adjust data quality thresholds if appropriate

3. Duplicate keys:

• Show sample duplicate values
• Options:
• Add deduplication logic (keep latest/first)
• Investigate source for duplicates
• Fail load and fix source
• Add DISTINCT or window function to transformation

4. Data quality rule failures:

• Show which rules failed and why
• Distinguish critical vs warning rules
• Options:
• Adjust rule thresholds (if too strict)
• Fix source data (if data is actually bad)
• Proceed with warnings (if non-critical)
• See data-quality-validation.md

Example Error Message to Present:

Data Quality Check Failed:
- Rule: IsPrimaryKey "order_id"
- Failure: Found 127 duplicate order_ids (0.5% of total rows)
- Sample duplicates: [10234, 10567, 10892, ...]

This could indicate:
1. Source data has duplicates (check data generation process)
2. Multiple loads without deduplication
3. Partition key included in order_id

Options:
1. Add deduplication keeping the latest record by timestamp
2. Investigate source system for root cause
3. Proceed with warning (not recommended for primary key)

How would you like to proceed?

Large Dataset Timeouts (Athena)

Symptoms:

• Athena query exceeds 30-minute timeout
• Query runs out of memory
• S3 read throttling errors

Root Causes:

• Dataset too large for single Athena query
• Insufficient Athena engine size
• Too many small files causing S3 throttling
• Complex transformations in single query

Solutions:

1. Break into batches:

• Split by date range or partition
• Load in multiple INSERT queries
• Example: Load one month at a time

2. Switch to Glue ETL:

• Glue can handle larger datasets with multiple workers
• Better for datasets > 1GB or millions of rows
• Provides better monitoring and retry logic
• See format-specific-loading.md for Glue examples

3. Increase Athena capacity:

• Use Athena v3 engine
• Increase DPU allocation in workgroup settings
• Consider Athena provisioned capacity for repeated large queries

4. Optimize file structure:

• Consolidate many small files (use Glue ETL)
• Use columnar formats (Parquet, ORC)
• Partition large datasets by date/region

Example Error Message to Present:

Athena Query Timeout:
Query exceeded 30-minute limit loading 5.2GB of data

Recommendations:
1. Switch to Glue ETL (recommended for datasets > 1GB)
   - Can handle 5.2GB with 5 G.1X workers in ~15 minutes
   - Better error handling and monitoring

2. Batch the load by date partition
   - Load 2024-01 through 2024-06 separately (6 queries)
   - Each query would handle ~850MB

Would you like me to:
A) Create a Glue ETL job for this load (recommended)
B) Set up batched Athena queries by month

Format-Specific Issues

CSV Parsing Errors

Symptoms:

• Columns shifted or misaligned
• Quoted values not parsed correctly
• Extra or missing columns

Solutions:

• Verify delimiter matches file (comma, tab, pipe)
• Set .option("quote", "\"") for quoted fields
• Set .option("escape", "\\") for escaped characters
• Use .option("mode", "DROPMALFORMED") to skip bad rows
• See format-specific-loading.md

JSON Parsing Errors

Symptoms:

• Multi-line JSON not parsing
• Nested structures flattened incorrectly
• Malformed JSON records causing failures

Solutions:

• Set .option("multiLine", "true") for multi-line objects
• Use .option("mode", "PERMISSIVE") to handle malformed records
• Check JSON schema matches expected structure
• Verify one JSON object per line for JSONL
• See format-specific-loading.md

Parquet Partition Issues

Symptoms:

• Partition columns not detected
• Schema evolution errors
• Missing partitions in results

Solutions:

• Verify Hive-style partitioning (key=value/)
• Use .option("mergeSchema", "true") for schema evolution
• Check partition column names match across files
• List S3 paths to confirm partition structure
• See format-specific-loading.md

Avro Library Errors

Symptoms:

• "Avro library not found" error
• Complex union types failing
• Schema registry connection errors

Solutions:

• Add --datalake-formats: iceberg,avro to Glue job arguments
• Or provide spark-avro JAR via --extra-jars
• Convert complex unions to STRING or handle with conditional logic
• See format-specific-loading.md

Error Severity Levels

Critical (Fail Immediately)

These errors should stop the workflow:

• IAM role doesn't exist or can't be assumed
• Source S3 path doesn't exist or is empty
• Target table exists with incompatible schema (cannot evolve)
• Primary key violations in data quality checks

Action: Present error clearly, provide remediation steps, wait for user action

Warnings (Proceed with Caution)

These issues should be flagged but allow continuation:

• High null percentage in optional columns
• Data quality warnings (not critical rules)
• Schema evolution needed (user approval required)
• Source files have malformed records (but most are valid)

Action: Show warning with details, ask user if they want to proceed

Informational

These are expected and don't require action:

• Using CLI fallback because MCP unavailable
• Sampling large files for schema inference
• Automatically inferring schema from source
• Creating IAM role because none exists

Action: Log for user visibility, proceed automatically

Troubleshooting Workflow

When encountering an error:

1. Identify the phase: Which workflow phase failed? 2. Read the error: Get full error message from CloudWatch/Athena 3. Check permissions: Verify IAM role has required policies 4. Validate data: Sample source data to check format/quality 5. Review configuration: Check Glue job args, Athena settings 6. Consult logs: Check CloudWatch logs for detailed stack traces 7. Search references: Check relevant reference doc for issue type

Getting Help

When presenting errors to users:

1. Be specific: Show exact error message and where it occurred 2. Provide context: What was being attempted when error happened 3. Offer solutions: Present 2-3 actionable options 4. Show impact: Explain what happens if user chooses each option 5. Ask clearly: Make the choice or next action explicit

Best Practices

1. Validate early: Check permissions and schema before starting load 2. Sample first: Test with small subset before full load 3. Monitor actively: Watch CloudWatch logs during execution 4. Handle gracefully: Don't let jobs fail silently - surface errors 5. Document issues: Keep track of common errors and solutions 6. Test transformations: Verify type casts and filters on sample data

Summary

Error handling workflow:

1. Detect error - Identify error type and severity 2. Diagnose root cause - Check logs, permissions, data 3. Present clearly - Show error and context to user 4. Offer solutions - Provide 2-3 actionable options 5. Execute fix - Apply chosen solution and retry 6. Validate resolution - Confirm error is resolved

With comprehensive error handling, the skill can guide users through issues confidently and get data loaded successfully.

Format-Specific Data Loading

Complete guide for reading and processing different file formats in Glue ETL jobs.

Overview

This reference covers format-specific configuration and code examples for loading data from various file formats into S3 Tables:

• CSV/TSV (delimited text files)
• JSON/JSONL (JavaScript Object Notation)
• Parquet (columnar format with embedded schema)
• Avro (row-based format with embedded schema)
• ORC (Optimized Row Columnar)

CSV and TSV Files

Basic CSV Reading

# CSV with custom delimiter
source_df = spark.read.format("csv") \
    .option("header", "true") \
    .option("delimiter", ",") \
    .option("inferSchema", "true") \
    .load(args['source_path'])

TSV (Tab-Separated Values)

# TSV (tab-separated)
source_df = spark.read.format("csv") \
    .option("header", "true") \
    .option("delimiter", "\t") \
    .load(args['source_path'])

CSV Options

Advanced CSV Example

# CSV with custom options
source_df = spark.read.format("csv") \
    .option("header", "true") \
    .option("delimiter", ",") \
    .option("quote", "\"") \
    .option("escape", "\\") \
    .option("nullValue", "NULL") \
    .option("dateFormat", "yyyy-MM-dd") \
    .option("timestampFormat", "yyyy-MM-dd HH:mm:ss") \
    .option("mode", "DROPMALFORMED") \
    .load(args['source_path'])

JSON and JSONL Files

JSON Lines (JSONL)

One JSON object per line (most common):

# JSON Lines (one JSON object per line)
source_df = spark.read.format("json").load(args['source_path'])

Nested JSON Handling

Option A: Flatten Nested Structures

from pyspark.sql.functions import col

# Flatten nested JSON
flattened_df = source_df.select(
    col("customer.customer_id").alias("customer_id"),
    col("customer.name").alias("customer_name"),
    col("customer.email").alias("email"),
    col("order_id"),
    col("order_date"),
    col("amount")
)

Option B: Preserve as STRUCT

No transformation needed - Iceberg supports STRUCT types:

# Preserve nested structure (no transformation)
# Schema becomes:
# - order_id: BIGINT
# - customer: STRUCT<customer_id:BIGINT, name:STRING, email:STRING>
# - order_date: DATE
# - amount: DECIMAL

JSON Options

Array Handling

# Explode array into separate rows
from pyspark.sql.functions import explode

df_with_items = source_df.select(
    col("order_id"),
    explode(col("items")).alias("item")
).select(
    col("order_id"),
    col("item.product_id"),
    col("item.quantity"),
    col("item.price")
)

# Or preserve as ARRAY type in Iceberg
# Schema: items ARRAY<STRUCT<product_id:STRING, quantity:INT, price:DECIMAL>>

Parquet Files

Basic Parquet Reading

# Parquet (direct read, schema preserved)
source_df = spark.read.format("parquet").load(args['source_path'])

Partitioned Parquet

Spark automatically detects Hive-style partitions:

# Partitioned Parquet (Spark auto-detects partitions)
source_df = spark.read.format("parquet").load("s3://bucket/events/")
# Partitions like year=2024/month=01/ are automatically handled

Detect Partition Structure

For partitioned data with Hive-style partitioning (e.g., year=2024/month=01/day=15/):

Using Python regex:

import re

# Example S3 path: s3://bucket/events/year=2024/month=01/day=15/part-0000.parquet
sample_s3_path = "s3://bucket/events/year=2024/month=01/day=15/part-0000.parquet"

# Extract partition key-value pairs
path_pattern = r'(\w+)=([^/]+)'
partitions = re.findall(path_pattern, sample_s3_path)
# Result: [('year', '2024'), ('month', '01'), ('day', '15')]

partition_columns = [col for col, _ in partitions]
print(f"Detected partition columns: {partition_columns}")
# Output: ['year', 'month', 'day']

Using AWS CLI:

# List S3 paths to identify partition patterns
aws s3 ls s3://bucket/events/ --recursive | head -20

# Look for patterns like:
# year=2024/month=01/day=01/
# year=2024/month=01/day=02/

Partition Column Inference

• Partition columns should typically be: INT, STRING, or DATE types
• Common partition patterns: year, month, day, region, category
• Important: Partition columns will NOT appear in the data files themselves (they're in the path)

Present Partition Info to User

Detected partitioned data structure:
- Partition columns: year (INT), month (INT), day (INT)
- Data columns: event_id, event_type, timestamp, user_id, properties
- Sample partition: year=2024/month=01/day=15
- Estimated partitions: ~90 (covering 3 months)

Avro Files

Basic Avro Reading

# Avro format
source_df = spark.read.format("avro").load(args['source_path'])

Avro Schema Extraction

Avro files contain embedded schemas. Extract and display:

Using Python avro library:

import avro.datafile
import avro.io
import json

# Read Avro file and extract schema
with open('downloaded-sample.avro', 'rb') as f:
    reader = avro.datafile.DataFileReader(f, avro.io.DatumReader())
    schema_json = reader.meta.get('avro.schema').decode('utf-8')
    schema = json.loads(schema_json)

    print("Avro Schema:")
    print(json.dumps(schema, indent=2))

    # Extract field names and types
    for field in schema['fields']:
        print(f"  {field['name']}: {field['type']}")

Using fastavro:

import fastavro

with open('downloaded-sample.avro', 'rb') as f:
    reader = fastavro.reader(f)
    schema = reader.writer_schema
    for field in schema['fields']:
        print(f"  {field['name']}: {field['type']}")

Avro to Iceberg Type Mapping

Handling Avro Union Types

Avro uses unions for nullable fields:

// Avro schema with nullable field
{
  "name": "age",
  "type": ["null", "int"]
}

Maps to Iceberg:

age INT  -- Nullable by default in Iceberg

For complex unions (non-nullable):

from pyspark.sql.functions import col, when

# Example: Handle union of int and string
df_with_union = source_df.withColumn(
    "age_clean",
    when(col("age").cast("int").isNotNull(), col("age").cast("int"))
    .otherwise(None)
)

Options for complex unions:

• Option A: Convert to JSON string and store as STRING
• Option B: Flatten union types into separate columns (age_int, age_string)
• Option C: Fail and ask user how to handle

Present Avro Schema to User

Detected Avro schema with 15 fields:
- user_id (long) → BIGINT
- username (string) → STRING
- age (union[null, int]) → INT (nullable)
- status (enum: active, inactive) → STRING
- metadata (map<string, string>) → MAP<STRING, STRING>
- preferences (record) → STRUCT

Glue Job Configuration for Avro

Option A: Use `--datalake-formats` (spark-avro built-in in Glue 5.1 or higher):

# In job DefaultArguments
'--datalake-formats': 'iceberg,delta,hudi,avro'

Option B: Provide spark-avro JAR:

# In create-job command
--default-arguments '{
  "--extra-jars": "s3://my-bucket/jars/spark-avro_2.12-3.4.0.jar"
}'

ORC Files

Basic ORC Reading

# ORC format
source_df = spark.read.format("orc").load(args['source_path'])

ORC files include embedded schema similar to Parquet. No special configuration needed.

Sampling Source Data

Before loading, sample source files to understand structure:

CSV Sampling

# Download and inspect first 10 lines
aws s3 cp s3://<bucket>/<key> - | head -10

Parquet Schema Inspection

import pyarrow.parquet as pq

# Read Parquet schema
table = pq.read_table('s3://<bucket>/<key>')
print(table.schema)

# Sample first 10 rows
df = table.to_pandas()
print(df.head(10))

JSON Sampling

# Download and inspect first 5 JSON objects
aws s3 cp s3://<bucket>/<key> - | head -5

Complete Glue ETL Script Template

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, ['JOB_NAME', 'source_path', 'target_table', 'source_format'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read source data based on format
if args['source_format'] == 'csv':
    source_df = spark.read.format("csv") \
        .option("header", "true") \
        .option("inferSchema", "true") \
        .load(args['source_path'])
elif args['source_format'] == 'json':
    source_df = spark.read.format("json").load(args['source_path'])
elif args['source_format'] == 'parquet':
    source_df = spark.read.format("parquet").load(args['source_path'])
elif args['source_format'] == 'avro':
    source_df = spark.read.format("avro").load(args['source_path'])
elif args['source_format'] == 'orc':
    source_df = spark.read.format("orc").load(args['source_path'])
else:
    raise ValueError(f"Unsupported format: {args['source_format']}")

# Apply transformations as needed
transformed_df = source_df.select(
    # Column transformations here
)

# Write to Iceberg table
transformed_df.writeTo(args['target_table']).append()

job.commit()

Format-Specific Common Issues

CSV Issues

Issue: Column type inference incorrect Solution: Explicitly specify schema or cast columns after reading

Issue: Quoted fields not parsed correctly Solution: Set .option("quote", "\"") and .option("escape", "\\")

JSON Issues

Issue: Multi-line JSON not parsing Solution: Set .option("multiLine", "true")

Issue: Malformed JSON records Solution: Set .option("mode", "DROPMALFORMED") or "PERMISSIVE"

Parquet Issues

Issue: Partition columns not detected Solution: Verify path follows Hive-style partitioning (key=value/)

Issue: Schema evolution errors Solution: Use .option("mergeSchema", "true") when reading

Avro Issues

Issue: Avro library not found Solution: Add --datalake-formats: iceberg,avro to job arguments

Issue: Complex union types failing Solution: Convert to STRING or handle with conditional logic

Best Practices

1. Always sample data first: Understand structure before loading 2. Validate schema mapping: Ensure source types map correctly to Iceberg 3. Handle malformed records: Use appropriate error handling mode 4. Test with small dataset: Verify transformations work before full load 5. Monitor CloudWatch logs: Check for parsing errors or warnings 6. Document format-specific options: Keep track of delimiter, quote char, etc. 7. Use schema evolution carefully: Understand impact on existing data

Summary

Different file formats require different reading configurations:

With format-specific configuration, Glue ETL can successfully load data from any supported format into S3 Tables.

Glue ETL Migration for Large Tables

Use Glue ETL (Path B) when Athena CTAS would time out, when transforms are complex, or when the migration needs to be scheduled/repeatable.

When to Use

• Source table over ~100 GB
• Complex column transformations that benefit from PySpark
• Migration needs to be scheduled or repeatable
• Source has more than 100 target partitions and batching is impractical

Job Setup

Requirements

• Glue 5.1 or higher (Spark 3.5.6, Iceberg 1.10.0)
• --datalake-formats iceberg job argument
• Catalog config in --conf job argument (not spark.conf.set()). See iceberg-catalog-config-and-usage.md for the exact keys.
• IAM role with S3 Tables, Glue, and S3 permissions

Job Configuration (JSON)

Use --cli-input-json to avoid shell escaping issues:

Glue --conf format: In Glue DefaultArguments, multiple Spark configs must be passed as a single --conf value with space-separated --conf key=value pairs. Do not split them into separate JSON keys — Glue only reads one --conf key.

{
    "Name": "migrate-to-s3tables",
    "Role": "arn:aws:iam::<account-id>:role/<glue-role>",
    "Command": {
        "Name": "glueetl",
        "ScriptLocation": "s3://<scripts-bucket>/scripts/migrate.py",
        "PythonVersion": "3"
    },
    "DefaultArguments": {
        "--datalake-formats": "iceberg",
        "--enable-glue-datacatalog": "true",
        "--conf": "<see iceberg-catalog-config-and-usage.md for S3 Tables Analytics Integration or REST config>"
    },
    "GlueVersion": "5.1",
    "NumberOfWorkers": 10,
    "WorkerType": "G.1X"
}

aws glue create-job --cli-input-json file://job-config.json --region <region>

Scale NumberOfWorkers based on source size: ~2 workers per 50 GB as a starting point.

PySpark Migration Script

import sys
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, [
    'JOB_NAME', 'source_database', 'source_table',
    'target_namespace', 'target_table'
])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read from source (Glue Data Catalog)
source_df = spark.read.table(
    f"glue_catalog.{args['source_database']}.{args['source_table']}"
)

# Apply transforms (customize as needed)
# Example: lowercase column names for S3 Tables compatibility
for col_name in source_df.columns:
    if col_name != col_name.lower():
        source_df = source_df.withColumnRenamed(col_name, col_name.lower())

# Write to S3 Table
target_table = f"s3tablescatalog.{args['target_namespace']}.{args['target_table']}"

source_df.writeTo(target_table) \
    .tableProperty("format-version", "2") \
    .createOrReplace()

# Verify row count
source_count = spark.read.table(
    f"glue_catalog.{args['source_database']}.{args['source_table']}"
).count()
target_count = spark.read.table(target_table).count()
print(f"Source rows: {source_count}, Target rows: {target_count}")

job.commit()

Key Points

• All catalog config goes in --conf job argument, never in spark.conf.set(). See iceberg-catalog-config-and-usage.md for the exact keys.
• No LOCATION clause -- S3 Tables manages storage
• Column names must be all lowercase for Athena visibility
• createOrReplace() handles both cases: creates the table if absent, replaces it if present (safe for re-runs)
• For partitioned writes, add .partitionedBy() before .createOrReplace()

Running and Monitoring

# Start the job
JOB_RUN_ID=$(aws glue start-job-run \
  --job-name "migrate-to-s3tables" \
  --arguments '{"--source_database":"legacy_db","--source_table":"orders","--target_namespace":"analytics","--target_table":"orders"}' \
  --query 'JobRunId' --output text)

# Check status
aws glue get-job-run --job-name "migrate-to-s3tables" --run-id "$JOB_RUN_ID"

Troubleshooting

Glue Job Configuration Guide

Guide for creating Glue jobs, configuring workers, advanced PySpark patterns, and monitoring for external data import pipelines.

Creating the Glue Job

Once you have the PySpark script saved to S3 (e.g., s3://<scripts-bucket>/glue-jobs/external-import-<table-name>.py), create the Glue job.

AWS CLI

aws glue create-job \
  --name "external-import-<source>-<table>" \
  --role "<glue-role-arn>" \
  --command "Name=glueetl,ScriptLocation=s3://<scripts-bucket>/glue-jobs/external-import-<table>.py,PythonVersion=3" \
  --connections "Connections=<glue-connection-name>" \
  --default-arguments '{
    "--datalake-formats": "iceberg",
    "--connection_name": "<glue-connection-name>",
    "--source_table": "<schema>.<table>",
    "--target_table": "<catalog>.<namespace>.<s3-table>",
    "--watermark_column": "<timestamp-column>",
    "--watermark_bucket": "<bucket>",
    "--watermark_key": "watermarks/<table-name>.txt",
    "--conf": "<see iceberg-catalog-config-and-usage.md for S3 Tables or standard Iceberg catalog config>",
    "--enable-glue-datacatalog": "true",
    "--enable-metrics": "true",
    "--enable-continuous-cloudwatch-log": "true"
  }' \
  --glue-version "5.1" \
  --number-of-workers 5 \
  --worker-type "G.1X" \
  --timeout 60 \
  --max-retries 1 \
  --region <region>

Job Configuration Parameters

Worker Types and Sizing

Choose worker type based on workload characteristics:

Number of workers guidance:

• Small tables (<1M rows, <1 GB): 2-5 workers, G.1X
• Medium tables (1M-10M rows, 1-10 GB): 5-10 workers, G.1X or G.2X
• Large tables (10M-100M rows, 10-100 GB): 10-20 workers, G.2X
• Very large tables (>100M rows, >100 GB): 20-50 workers, G.2X or G.4X

Start conservative and scale up based on job duration and throughput.

Timeout Configuration

Set timeout based on expected job duration:

• Small incremental loads: 15-30 minutes
• Medium incremental loads: 30-60 minutes
• Large incremental loads: 60-120 minutes
• Full refresh of large tables: 120-480 minutes

Add buffer for source database query time and network latency.

Retry Configuration

Configure retries for transient failures:

'MaxRetries': 1  # Retry once on failure

For production pipelines, consider:

• Setting MaxRetries to 1-2 for transient network issues
• Using Glue job bookmarks to avoid duplicate processing
• Implementing idempotent logic (upsert instead of append)

Important Job Arguments

Required arguments:

• --datalake-formats iceberg: Required for S3 Tables and standard Iceberg targets
• --enable-glue-datacatalog: Enable Glue Data Catalog integration for Iceberg
• --conf: Spark catalog configuration. See iceberg-catalog-config-and-usage.md for the exact keys per target type.
• --enable-metrics: Publish CloudWatch metrics
• --enable-continuous-cloudwatch-log: Stream logs to CloudWatch

Optional arguments:

• --enable-spark-ui: Enable Spark UI for debugging (requires S3 bucket)
• --spark-event-logs-path: Where to store Spark UI logs
• --conf spark.sql.adaptive.enabled=true: Enable adaptive query execution
• --conf spark.sql.adaptive.coalescePartitions.enabled=true: Optimize partition count

Network Configuration

If the source database is in a VPC, ensure the Glue job has network access:

'Connections': {
    'Connections': ['<glue-connection-name>']
}

The connection specifies:

• VPC
• Subnet
• Security groups
• Availability zone

Glue provisions ENIs in the specified subnet to access the database.

Advanced PySpark Patterns

Parallel Reads with Partitioning

For large tables, read data in parallel using Spark partitioning:

# Read with parallel partitions
source_df = spark.read.format("jdbc").options(
    url=jdbc_url,
    dbtable="large_table",
    numPartitions=10,  # Read with 10 parallel connections
    partitionColumn="id",  # Partition on this column
    lowerBound=1,  # Min value
    upperBound=10000000  # Max value
).load()

This creates 10 parallel queries:

• Partition 1: WHERE id >= 1 AND id < 1000000
• Partition 2: WHERE id >= 1000000 AND id < 2000000
• ...
• Partition 10: WHERE id >= 9000000 AND id <= 10000000

Best practices:

• Use a numeric column with even distribution
• Set numPartitions = number of workers × cores per worker
• Choose lowerBound and upperBound based on actual data range

Deduplication Logic

If there's risk of duplicate records (job retries, late arrivals):

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

# Deduplicate by primary key, keeping latest by watermark
window = Window.partitionBy("primary_key").orderBy(col(watermark_column).desc())
deduplicated_df = source_df.withColumn("row_num", row_number().over(window)) \
    .filter(col("row_num") == 1) \
    .drop("row_num")

Type Conversion and Validation

Add data quality checks and type conversions:

from pyspark.sql.functions import col, when

transformed_df = source_df.select(
    # Safe type casting with null handling
    when(col("amount").cast("double").isNotNull(), col("amount").cast("double"))
        .otherwise(0.0).alias("amount"),

    # String trimming and validation
    when(col("email").rlike(r"^[\w\.-]+@[\w\.-]+\.\w+$"), col("email"))
        .otherwise(None).alias("email"),

    # Date parsing with fallback
    when(col("order_date").isNotNull(),
         to_date(col("order_date"), "yyyy-MM-dd"))
        .otherwise(None).alias("order_date")
)

Watermark with Buffer for Late Arrivals

If source data can arrive late (event timestamp < updated timestamp):

from datetime import timedelta

# Load data from 1 day before last watermark to catch late arrivals
buffer_watermark = (datetime.strptime(last_watermark, '%Y-%m-%d %H:%M:%S')
                    - timedelta(days=1)).strftime('%Y-%m-%d %H:%M:%S')

filtered_df = source_df.filter(
    f"{args['watermark_column']} > '{buffer_watermark}'"
)

# Then use upsert to avoid duplicates

Monitoring and Observability

CloudWatch Logs

Glue streams job logs to CloudWatch Logs under:

• Log group: /aws-glue/jobs/output
• Log stream: <job-name>-<job-run-id>

Key log patterns to monitor:

• Last watermark: <value> - Starting point for incremental load
• Loading X new/updated records - How many records found
• Updated watermark to: <value> - New watermark after load
• ERROR - Any errors during execution

CloudWatch Metrics

With --enable-metrics, Glue publishes:

• glue.driver.aggregate.numCompletedTasks - Tasks completed
• glue.driver.aggregate.elapsedTime - Job duration
• glue.driver.aggregate.recordsRead - Records read from source
• glue.driver.aggregate.bytesRead - Bytes read from source

Set up CloudWatch alarms for:

• Job failures (state = FAILED)
• Long-running jobs (duration > threshold)
• No records loaded (might indicate source issue)

Spark UI

Enable Spark UI for detailed execution metrics:

'DefaultArguments': {
    '--enable-spark-ui': 'true',
    '--spark-event-logs-path': 's3://<logs-bucket>/spark-logs/'
}

Access via Glue console → Job runs → View Spark UI

Use Spark UI to:

• Identify slow stages (data skew, shuffle issues)
• Analyze task distribution across workers
• Debug memory issues (GC time, spills to disk)

Script Storage and Versioning

Best practices for script management:

1. Store scripts in S3: s3://<scripts-bucket>/glue-jobs/<job-name>.py 2. Version scripts: Use S3 versioning or include version in filename 3. Separate environments: Different buckets for dev/staging/prod 4. Use Git: Maintain scripts in Git, deploy to S3 via CI/CD

Example structure:

s3://my-glue-scripts/
  prod/
    external-import-customers.py
    external-import-orders.py
  dev/
    external-import-customers.py
    external-import-orders.py

Testing Scripts Locally

Test PySpark scripts locally before deploying to Glue:

# Install dependencies
pip install pyspark boto3

# Run script locally (modify to use local Spark)
python external-import-customers.py \
  --JOB_NAME test-run \
  --connection_name test-connection \
  --source_table customers \
  --target_table local.test.customers \
  --watermark_column updated_at \
  --watermark_bucket test-bucket \
  --watermark_key watermarks/customers.txt

For full local testing, use AWS Glue Docker images:

docker pull amazon/aws-glue-libs:glue_libs_5.0.0_image_01

Summary

Glue ETL job creation workflow:

1. Choose template - Append, Upsert, Custom SQL, or Full Refresh 2. Customize script - Add transformations, validation, error handling 3. Save to S3 - Store script in versioned S3 location 4. Create job - Use MCP or CLI with appropriate configuration 5. Size workers - Choose worker type and count based on data volume 6. Configure monitoring - Enable CloudWatch logs and metrics 7. Test locally - Validate logic before deploying (optional)

With a well-configured Glue job, external database data flows continuously into S3 Tables with minimal operational overhead.

Glue ETL Job Creation Guide

Complete guide for creating AWS Glue ETL jobs that import data from external databases into S3 Tables.

Overview

Glue ETL jobs use PySpark to connect to external databases via connections, read data incrementally using watermark columns, apply transformations, and write to Iceberg tables in S3 Tables.

PySpark Script Structure

Basic Incremental Append Template

For immutable data (transactions, events, logs) where you only need to append new records:

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame
import boto3
from datetime import datetime
from pyspark.sql.functions import lit

# Parse job arguments
args = getResolvedOptions(sys.argv, [
    'JOB_NAME',
    'connection_name',
    'source_table',
    'target_table',
    'watermark_column',
    'watermark_bucket',
    'watermark_key'
])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read last watermark from S3
s3 = boto3.client('s3')
try:
    obj = s3.get_object(Bucket=args['watermark_bucket'], Key=args['watermark_key'])
    last_watermark = obj['Body'].read().decode('utf-8').strip()
    print(f"Last watermark: {last_watermark}")
except s3.exceptions.NoSuchKey:
    last_watermark = '1970-01-01 00:00:00'  # Default for timestamp
    # OR last_watermark = '0'  # Default for ID column
    print("No previous watermark found, starting from beginning")

# Read from external database using Glue connection
source_df = glueContext.create_dynamic_frame.from_catalog(
    database="<temp-catalog-db>",
    table_name="<source-table>",
    transformation_ctx="source_df",
    additional_options={
        "connectionName": args['connection_name']
    }
).toDF()

# Apply incremental filter
filtered_df = source_df.filter(
    f"{args['watermark_column']} > '{last_watermark}'"
)

row_count = filtered_df.count()
print(f"Loading {row_count} new/updated records")

if row_count > 0:
    # Apply transformations (type casting, column mapping, etc.)
    transformed_df = filtered_df.select(
        # Map source columns to target schema
        filtered_df["source_col1"].cast("int").alias("target_col1"),
        filtered_df["source_col2"].alias("target_col2"),
        filtered_df["source_col3"].cast("double").alias("target_col3"),
        # Add load metadata
        lit(datetime.now()).alias("load_timestamp")
    )

    # Write to Iceberg table (append mode)
    transformed_df.writeTo(args['target_table']).append()

    # Update watermark in S3
    new_watermark = filtered_df.agg({args['watermark_column']: "max"}).collect()[0][0]
    s3.put_object(
        Bucket=args['watermark_bucket'],
        Key=args['watermark_key'],
        Body=str(new_watermark)
    )
    print(f"Updated watermark to: {new_watermark}")
    print(f"Successfully loaded {row_count} records")
else:
    print("No new records to load")

job.commit()

Incremental Upsert Template

For mutable data (customer profiles, product catalog) where records can be updated:

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import col, lit
import boto3
from datetime import datetime

# Parse job arguments
args = getResolvedOptions(sys.argv, [
    'JOB_NAME',
    'connection_name',
    'source_table',
    'target_table',
    'watermark_column',
    'primary_key',  # Column used for merging
    'watermark_bucket',
    'watermark_key'
])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read last watermark
s3 = boto3.client('s3')
try:
    obj = s3.get_object(Bucket=args['watermark_bucket'], Key=args['watermark_key'])
    last_watermark = obj['Body'].read().decode('utf-8').strip()
    print(f"Last watermark: {last_watermark}")
except s3.exceptions.NoSuchKey:
    last_watermark = '1970-01-01 00:00:00'
    print("No previous watermark found, starting from beginning")

# Read from external database
source_df = glueContext.create_dynamic_frame.from_catalog(
    database="<temp-catalog-db>",
    table_name="<source-table>",
    transformation_ctx="source_df",
    additional_options={
        "connectionName": args['connection_name']
    }
).toDF()

# Get new/updated records
changed_records_df = source_df.filter(
    f"{args['watermark_column']} > '{last_watermark}'"
)

row_count = changed_records_df.count()
print(f"Found {row_count} new/updated records")

if row_count > 0:
    # Apply transformations
    transformed_df = changed_records_df.select(
        changed_records_df["customer_id"].cast("int").alias("customer_id"),
        changed_records_df["customer_name"].alias("name"),
        changed_records_df["email"].alias("email"),
        changed_records_df["status"].alias("status"),
        changed_records_df["updated_at"].alias("updated_at"),
        lit(datetime.now()).alias("load_timestamp")
    )

    # Create temporary view for MERGE operation
    transformed_df.createOrReplaceTempView("source_view")

    # Execute MERGE INTO (upsert)
    spark.sql(f"""
    MERGE INTO {args['target_table']} AS target
    USING source_view AS source
    ON target.{args['primary_key']} = source.{args['primary_key']}
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
    """)

    # Update watermark
    new_watermark = changed_records_df.agg({args['watermark_column']: "max"}).collect()[0][0]
    s3.put_object(
        Bucket=args['watermark_bucket'],
        Key=args['watermark_key'],
        Body=str(new_watermark)
    )
    print(f"Updated watermark to: {new_watermark}")
    print(f"Upserted {row_count} records")
else:
    print("No new records to process")

job.commit()

Custom SQL Query Template

When users want to filter or transform at source with custom SQL:

import sys
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import lit
import boto3
from datetime import datetime

# Parse job arguments
args = getResolvedOptions(sys.argv, [
    'JOB_NAME',
    'connection_name',
    'source_query',  # SQL query to execute
    'target_table',
    'watermark_column',
    'watermark_bucket',
    'watermark_key',
    'jdbc_driver'
])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Retrieve JDBC credentials from Glue connection
jdbc_conf = glueContext.extract_jdbc_conf(args['connection_name'])

# Read last watermark
s3 = boto3.client('s3')
try:
    obj = s3.get_object(Bucket=args['watermark_bucket'], Key=args['watermark_key'])
    last_watermark = obj['Body'].read().decode('utf-8').strip()
    print(f"Last watermark: {last_watermark}")
except s3.exceptions.NoSuchKey:
    last_watermark = '1970-01-01 00:00:00'
    print("Starting from beginning")

# Build query with watermark filter
query = f"""
SELECT * FROM ({args['source_query']}) AS base_query
WHERE {args['watermark_column']} > '{last_watermark}'
"""

print(f"Executing query: {query}")

# Read using JDBC with custom query
source_df = spark.read.format("jdbc").options(
    url=jdbc_conf['url'],
    dbtable=f"({query}) AS subquery",
    user=jdbc_conf['user'],
    password=jdbc_conf['password'],
    driver=args['jdbc_driver']  # e.g., "oracle.jdbc.OracleDriver"
).load()

row_count = source_df.count()
print(f"Query returned {row_count} records")

if row_count > 0:
    # Add load metadata
    transformed_df = source_df.withColumn("load_timestamp", lit(datetime.now()))

    # Write to Iceberg table
    transformed_df.writeTo(args['target_table']).append()

    # Update watermark
    new_watermark = source_df.agg({args['watermark_column']: "max"}).collect()[0][0]
    s3.put_object(
        Bucket=args['watermark_bucket'],
        Key=args['watermark_key'],
        Body=str(new_watermark)
    )
    print(f"Updated watermark to: {new_watermark}")
else:
    print("No new records")

job.commit()

Full Refresh Template

For small dimension tables or when source doesn't support watermarks:

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import lit
from datetime import datetime

# Parse job arguments
args = getResolvedOptions(sys.argv, [
    'JOB_NAME',
    'connection_name',
    'source_table',
    'target_table'
])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read all records from source
source_df = glueContext.create_dynamic_frame.from_catalog(
    database="<temp-catalog-db>",
    table_name="<source-table>",
    transformation_ctx="source_df",
    additional_options={
        "connectionName": args['connection_name']
    }
).toDF()

row_count = source_df.count()
print(f"Loading {row_count} records (full refresh)")

# Apply transformations
transformed_df = source_df.select(
    source_df["col1"].alias("col1"),
    source_df["col2"].alias("col2"),
    lit(datetime.now()).alias("load_timestamp")
)

# Overwrite target table
transformed_df.writeTo(args['target_table']).overwritePartitions()

print(f"Full refresh completed: {row_count} records loaded")

job.commit()

Iceberg Catalog Config and Engine Access Patterns

How to configure Spark catalog settings, select a target format, and address tables from each engine.

S3 Tables (Default)

Managed Iceberg tables with automatic compaction, snapshot management, and multi-engine access.

• Catalog path: The table bucket is configured in --conf via glue.id, so the write path is 3-part: s3tablescatalog.<namespace>.<table>
• No LOCATION clause in CREATE TABLE
• Table and column names must be lowercase
• Requires Glue 5.1 or higher and --datalake-formats iceberg job argument
• All spark.sql.catalog.* config goes in --conf job arguments, never in spark.conf.set() (Glue 5.x static config restriction)
• Delegate table creation to creating-data-lake-table

Two access methods exist. Use Analytics Integration when the table needs to be visible to Athena, Redshift, or EMR. Use REST Endpoint when only Glue Spark jobs access the table.

Analytics Integration (recommended for multi-engine access):

spark.sql.catalog.s3tablescatalog=org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.s3tablescatalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog
spark.sql.catalog.s3tablescatalog.glue.id=<account-id>:s3tablescatalog/<table-bucket-name>
spark.sql.catalog.s3tablescatalog.warehouse=<table-bucket-arn>

The warehouse parameter is required. Without it Spark fails with "Cannot derive default warehouse location".

REST Endpoint (Glue-only access):

spark.sql.catalog.s3tables=org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.s3tables.type=rest
spark.sql.catalog.s3tables.uri=https://s3tables.<region>.amazonaws.com/iceberg
spark.sql.catalog.s3tables.warehouse=<table-bucket-arn>
spark.sql.catalog.s3tables.rest.sigv4-enabled=true
spark.sql.catalog.s3tables.rest.signing-name=s3tables
spark.sql.catalog.s3tables.rest.signing-region=<region>
spark.sql.catalog.s3tables.io-impl=org.apache.iceberg.aws.s3.S3FileIO

Tables created via REST are NOT visible in Athena or Redshift.

`--conf` format in Glue DefaultArguments: Pass as a single string. First pair has no --conf prefix; subsequent pairs are space-separated with --conf prefix:

"--conf": "spark.sql.catalog.s3tablescatalog=org.apache.iceberg.spark.SparkCatalog --conf spark.sql.catalog.s3tablescatalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog --conf spark.sql.catalog.s3tablescatalog.glue.id=<account-id>:s3tablescatalog/<table-bucket-name> --conf spark.sql.catalog.s3tablescatalog.warehouse=<table-bucket-arn>"

Use --cli-input-json file://config.json to avoid shell escaping issues.

Write path (PySpark):

df.writeTo("s3tablescatalog.<namespace>.<table>").append()

Standard Iceberg on General Purpose Bucket

Self-managed Iceberg tables on regular S3 buckets. User handles compaction and snapshot cleanup.

• Catalog path: glue_catalog.<database>.<table> (via Glue Data Catalog)
• LOCATION clause IS required: LOCATION 's3://<bucket>/<prefix>/'
• Registered in Glue Data Catalog as normal
• Works with Glue 5.1 or higher and --datalake-formats iceberg job argument
• All spark.sql.catalog.* config goes in --conf job arguments, never in spark.conf.set()

Glue job catalog config:

spark.sql.catalog.glue_catalog=org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.glue_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog
spark.sql.catalog.glue_catalog.warehouse=s3://<bucket>/<warehouse-prefix>/

The warehouse parameter sets the default base path for new tables.

Write path (PySpark):

df.writeTo("glue_catalog.<database>.<table>").append()

Athena DDL:

CREATE TABLE <database>.<table> (
  col1 STRING,
  col2 INT
)
LOCATION 's3://<bucket>/<prefix>/'
TBLPROPERTIES ('table_type' = 'ICEBERG')

Parquet / ORC / CSV on S3

Raw files written to S3 with no Iceberg table metadata. Queryable via external tables in Athena.

• No table management (no compaction, no snapshots, no schema evolution)
• User must create an external table in Glue catalog to query with Athena
• Suitable when the user explicitly wants raw files, not a managed table

Write path (PySpark):

# Parquet
df.write.format("parquet").mode("overwrite").save("s3://<bucket>/<prefix>/")

# ORC
df.write.format("orc").mode("overwrite").save("s3://<bucket>/<prefix>/")

# CSV
df.write.format("csv").option("header", "true").mode("overwrite").save("s3://<bucket>/<prefix>/")

External table for querying:

CREATE EXTERNAL TABLE <database>.<table> (
  col1 STRING,
  col2 INT
)
STORED AS PARQUET
LOCATION 's3://<bucket>/<prefix>/'

Gotchas

• S3 Tables CREATE TABLE must NOT include a LOCATION clause. Standard Iceberg MUST include one.
• The s3tablescatalog federated catalog uses slash-separated paths in Athena: "s3tablescatalog/<bucket>"."<namespace>"."<table>". Spark uses dot-separated: s3tablescatalog.<namespace>.<table> (the bucket is configured in --conf via glue.id).
• Parquet/ORC/CSV targets do not create Iceberg metadata -- they are raw files only. No schema evolution, time travel, or ACID transactions.
• Discover available MCP tools by keyword search -- do not hardcode tool names.

Engine Access Patterns

How each engine reads and writes to each target format. Use this when building jobs that read from one format and write to another, or when validating ingested data.

S3 Tables

Spark with Analytics Integration and Athena both use 4 levels, but Athena uses slash-separated catalog paths while Spark uses dots. Spark with REST uses 3 levels because the table bucket is embedded in the --conf warehouse ARN.

Standard Iceberg

Standard Iceberg tables are registered in the default Glue Data Catalog. Athena queries them without a catalog prefix. Spark requires the catalog name from --conf (e.g., glue_catalog).

Parquet / ORC / CSV

No catalog registration needed for Spark reads — point directly at the S3 path. Athena requires an external table definition in the Glue catalog.

Decision Guide

Incremental Loading Strategies

Complete guide for configuring incremental data loading from external databases.

Overview

Incremental loading imports only new or changed records instead of the entire dataset on each run. This is essential for recurring pipelines to minimize data transfer and processing time.

Identify Watermark Column

A watermark column tracks which records have been loaded. The Glue job queries for records where watermark > last_loaded_value.

Common Watermark Patterns

Timestamp column (preferred):

• updated_at, modified_date, last_changed, etl_timestamp
• Query: WHERE timestamp_col > '2024-03-12 10:30:00'
• Best for: Mutable data that gets updated

Monotonic ID column:

• id, order_id, transaction_id (auto-incrementing)
• Query: WHERE id > 1234567
• Best for: Immutable data with sequential IDs

Both timestamp and ID:

• Use timestamp for recent changes, ID as fallback for historical data
• Query: WHERE timestamp_col > '...' OR (timestamp_col IS NULL AND id > ...)

Ask the User

Present candidates from the source schema:

I found these potential watermark columns:
1. CREATED_DATE (TIMESTAMP) - Never changes once set
2. UPDATED_AT (TIMESTAMP) - Updates when record changes (recommended)
3. ID (NUMBER) - Auto-incrementing primary key

Which should I use to track new/updated records?

Recommendation logic:

• If updated_at or modified_date exists → Recommend this (captures updates)
• Else if timestamp column exists → Use creation timestamp
• Else if auto-incrementing ID → Use ID
• Else → Recommend full refresh

Determine Load Strategy

Incremental Append (New Records Only)

Best for: Immutable data

• Transaction logs
• Event streams
• Historical orders
• Audit trails

How it works:

1. Query source for records where watermark > last_watermark 2. Append new records to target table 3. Update watermark to max value from current batch

Pros: Simple, fast, no deduplication needed Cons: Doesn't capture updates to existing records

PySpark example:

# Filter for new records
new_records_df = source_df.filter(
    f"{watermark_column} > '{last_watermark}'"
)

# Append to target
new_records_df.writeTo(target_table).append()

Incremental Upsert (New + Updated Records)

Best for: Mutable data

• Customer profiles
• Product catalogs
• Employee records
• Account balances

How it works:

1. Query source for records where watermark > last_watermark 2. Merge into target table using primary key 3. Update existing records, insert new ones 4. Update watermark

Pros: Captures both new records and updates Cons: More complex, requires MERGE operation

PySpark example:

# Get new/updated records
changed_records_df = source_df.filter(
    f"{watermark_column} > '{last_watermark}'"
)

# Merge into target (upsert)
spark.sql(f"""
MERGE INTO {target_table} AS target
USING changed_records AS source
ON target.{primary_key} = source.{primary_key}
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
""")

Full Refresh

Best for:

• Small dimension tables (< 10K rows)
• Data without watermark columns
• When source doesn't support incremental queries

How it works:

1. Truncate or drop target table 2. Load all records from source 3. No watermark needed

Pros: Simple, guarantees data consistency Cons: Inefficient for large tables, higher data transfer costs

PySpark example:

# Read all records
all_records_df = source_df.select("*")

# Overwrite target table
all_records_df.writeTo(target_table).overwritePartitions()

Watermark Storage Options

The Glue job needs to persist the last loaded watermark value between runs.

Option A: S3 File (Simple)

Store watermark in a text file in S3.

Advantages:

• Simple to implement
• No additional AWS services
• Easy to inspect and debug

Implementation:

import boto3

s3 = boto3.client('s3')
watermark_bucket = args['watermark_bucket']
watermark_key = args['watermark_key']

# Read last watermark
try:
    obj = s3.get_object(Bucket=watermark_bucket, Key=watermark_key)
    last_watermark = obj['Body'].read().decode('utf-8').strip()
    print(f"Last watermark: {last_watermark}")
except s3.exceptions.NoSuchKey:
    last_watermark = '1970-01-01 00:00:00'  # Default for timestamp
    # OR last_watermark = '0'  # Default for ID
    print("No previous watermark found, starting from beginning")

# After loading, update watermark
new_watermark = filtered_df.agg({watermark_column: "max"}).collect()[0][0]
s3.put_object(
    Bucket=watermark_bucket,
    Key=watermark_key,
    Body=str(new_watermark)
)
print(f"Updated watermark to: {new_watermark}")

S3 path structure:

s3://my-glue-watermarks/
  customers.txt          → "2024-03-12 14:30:00"
  orders.txt             → "2024-03-12 14:25:00"
  products.txt           → "2024-03-10 08:00:00"

Option B: DynamoDB Table (Robust)

Store watermarks in a DynamoDB table with one item per job.

Advantages:

• Atomic updates
• Query watermarks programmatically
• Can store additional metadata (last run time, row count, etc.)

Create table:

aws dynamodb create-table \
  --table-name glue-job-watermarks \
  --attribute-definitions \
    AttributeName=job_name,AttributeType=S \
  --key-schema \
    AttributeName=job_name,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --region <region>

Implementation:

import boto3
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('glue-job-watermarks')
job_name = args['JOB_NAME']

# Read last watermark
try:
    response = table.get_item(Key={'job_name': job_name})
    item = response['Item']
    last_watermark = item['watermark']
    print(f"Last watermark for {job_name}: {last_watermark}")
except KeyError:
    last_watermark = '1970-01-01 00:00:00'
    print("No previous watermark found, starting from beginning")

# After loading, update watermark
new_watermark = filtered_df.agg({watermark_column: "max"}).collect()[0][0]
table.put_item(Item={
    'job_name': job_name,
    'watermark': str(new_watermark),
    'last_run_time': datetime.now().isoformat(),
    'rows_loaded': row_count
})
print(f"Updated watermark to: {new_watermark}")

Option C: Query Target Table (Advanced)

Query the target S3 Table to determine the max watermark value.

Advantages:

• No external storage needed
• Watermark always matches actual data

Disadvantages:

• Requires target table scan (can be slow)
• Doesn't work for first run (empty table)

Implementation:

# Query target table for max watermark
try:
    max_watermark_df = spark.sql(f"""
        SELECT MAX({watermark_column}) as max_value
        FROM {target_table}
    """)
    last_watermark = max_watermark_df.collect()[0]['max_value']
    if last_watermark is None:
        last_watermark = '1970-01-01 00:00:00'
    print(f"Max watermark in target: {last_watermark}")
except:
    last_watermark = '1970-01-01 00:00:00'
    print("Target table empty or doesn't exist, starting from beginning")

Recommendation: Use Option A (S3 file) for simplicity unless you have specific requirements for DynamoDB's features.

Handling Edge Cases

Timezone Considerations

Problem: Source database uses one timezone, target uses another Solution: Normalize all timestamps to UTC

from pyspark.sql.functions import to_utc_timestamp

# Convert source timestamp to UTC
df_utc = source_df.withColumn(
    "timestamp_utc",
    to_utc_timestamp(col("source_timestamp"), "America/New_York")
)

Backfill Historical Data

Scenario: Need to load historical data before starting incremental loads

Approach:

1. Set watermark to earliest desired date: 1900-01-01 00:00:00 2. Run job once to load all historical data 3. Subsequent runs will be incremental from that point forward

OR load in batches:

# Batch 1: Load 2020 data
WHERE timestamp >= '2020-01-01' AND timestamp < '2021-01-01'

# Batch 2: Load 2021 data
WHERE timestamp >= '2021-01-01' AND timestamp < '2022-01-01'

# Batch 3: Load 2022+ data
WHERE timestamp >= '2022-01-01'

# Then switch to incremental

Late-Arriving Data

Problem: Records arrive after their timestamp (e.g., event from yesterday arrives today)

Solution 1: Add buffer window

# Load data from 1 day before last watermark to catch late arrivals
buffer_watermark = last_watermark - timedelta(days=1)
WHERE timestamp > buffer_watermark

Solution 2: Use separate updated_at column

# Use updated_at instead of event_timestamp
WHERE updated_at > last_watermark

Deleted Records

Problem: Source deletes records, but incremental load doesn't capture deletions

Solutions:

Option 1: Periodic full refresh

• Run incremental loads daily
• Run full refresh weekly to remove deleted records

Option 2: Soft deletes

• Source system marks records as deleted instead of removing them
• Filter: WHERE updated_at > last_watermark OR deleted_at > last_watermark

Option 3: Compare and prune

• Periodically query source for all IDs
• Find IDs in target that don't exist in source
• Delete those records from target

Duplicate Records

Problem: Same record loaded multiple times due to job retries or watermark issues

Prevention:

1. Use upsert instead of append for mutable data 2. Add deduplication logic:

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

# Deduplicate by primary key, keeping latest by watermark
window = Window.partitionBy("primary_key").orderBy(col(watermark_column).desc())
deduplicated_df = df.withColumn("row_num", row_number().over(window)) \
    .filter(col("row_num") == 1) \
    .drop("row_num")

Performance Optimization

Index Watermark Column

Ensure the watermark column has an index in the source database:

-- Oracle
CREATE INDEX idx_customers_updated_at ON CUSTOMERS(UPDATED_AT);

-- SQL Server
CREATE INDEX idx_customers_updated_at ON CUSTOMERS(UPDATED_AT);

-- PostgreSQL
CREATE INDEX idx_customers_updated_at ON customers(updated_at);

Without an index, source database will do full table scans.

Batch Size Tuning

For high-volume tables, load data in smaller batches:

# Load 1 hour of data at a time
batch_size = timedelta(hours=1)
current_watermark = last_watermark

while current_watermark < datetime.now():
    next_watermark = current_watermark + batch_size

    batch_df = source_df.filter(
        (col(watermark_column) > current_watermark) &
        (col(watermark_column) <= next_watermark)
    )

    batch_df.writeTo(target_table).append()

    current_watermark = next_watermark

Parallel Reads

Use Spark's partitioning for parallel reads from source:

source_df = spark.read.format("jdbc").options(
    url=jdbc_url,
    dbtable=table_name,
    numPartitions=10,  # Read in parallel with 10 partitions
    partitionColumn=watermark_column,
    lowerBound=last_watermark,
    upperBound=current_time
).load()

Monitoring and Alerting

Track these metrics for each incremental load:

• Rows loaded: Number of new/updated records
• Watermark advancement: How much watermark advanced
• Load duration: Time taken for the job
• Data lag: Difference between source max watermark and loaded watermark

# Log metrics
print(f"Job metrics:")
print(f"  Rows loaded: {row_count}")
print(f"  Previous watermark: {last_watermark}")
print(f"  New watermark: {new_watermark}")
print(f"  Watermark advancement: {new_watermark - last_watermark}")
print(f"  Load duration: {load_duration} seconds")

# Publish to CloudWatch (optional)
cloudwatch = boto3.client('cloudwatch')
cloudwatch.put_metric_data(
    Namespace='GlueJobs',
    MetricData=[{
        'MetricName': 'RowsLoaded',
        'Value': row_count,
        'Unit': 'Count',
        'Dimensions': [{'Name': 'JobName', 'Value': job_name}]
    }]
)

Best Practices

1. Choose the right watermark column: Prefer updated_at over created_at for mutable data 2. Test with small batches first: Verify logic before full-scale loads 3. Add buffer for late arrivals: Consider loading data from 1 day before watermark 4. Monitor watermark advancement: Alert if watermark stops advancing 5. Handle timezones consistently: Convert all timestamps to UTC 6. Index watermark column in source: Dramatically improves query performance 7. Use upsert for mutable data: Prevents duplicates and captures updates 8. Store watermark reliably: S3 file is simple and sufficient for most cases

Summary

Incremental loading workflow:

1. Identify watermark column - Timestamp or auto-incrementing ID 2. Choose load strategy - Append (immutable) vs Upsert (mutable) vs Full Refresh 3. Store watermark - S3 file, DynamoDB, or query target table 4. Handle edge cases - Timezones, late arrivals, deletions, duplicates 5. Optimize performance - Index watermark, batch loading, parallel reads 6. Monitor - Track rows loaded, watermark advancement, data lag

With proper incremental loading, recurring pipelines efficiently sync only changed data from external databases.

JDBC Database Ingest

Move data from a JDBC source (Oracle, SQL Server, PostgreSQL, MySQL, RDS, Aurora, Redshift) into the data lake. Assumes a Glue connection exists. If it doesn't, delegate to the connecting-to-data-source skill first.

Contents

• Prerequisites
• Workflow
• Parallel Reads
• Type Mapping
• Connection Errors

Prerequisites

• A tested Glue connection (created via connecting-to-data-source skill)
• Source table name, schema, and optional filter SQL
• Target table (existing or to be created via creating-data-lake-table skill)
• Target format decided (default S3 Tables; see iceberg-catalog-config-and-usage.md)

Workflow

1. Confirm connection exists

aws glue get-connection --name <CONNECTION_NAME> --region <REGION>

If the connection does not exist, stop and delegate to connecting-to-data-source.

2. Identify source scope

Ask the user which tables, views, or custom SQL query. See jdbc-schema-discovery.md for crawler-based discovery, direct schema inspection, and custom SQL patterns.

3. Decide load strategy

4. Create target table if needed

If the target table doesn't exist, delegate to creating-data-lake-table. Never create it inline.

5. Build the Glue 5.1 or higher job

Use the PySpark templates in glue-job-scripts.md and the job config guidance in glue-job-config.md.

Reference the Glue connection via job Connections property:

"Connections": {"Connections": ["<CONNECTION_NAME>"]}

In the script, read via connection name -- no credentials in code:

source_df = glueContext.create_dynamic_frame.from_options(
    connection_type="jdbc",
    connection_options={
        "useConnectionProperties": "true",
        "connectionName": args['connection_name'],
        "dbtable": args['source_table']
    }
).toDF()

6. Test, validate, schedule

• Run the job manually once
• Validate per data-quality-validation.md: row counts, null checks on critical columns, spot-check samples
• For recurring pipelines, create a Glue Trigger per testing-and-scheduling.md

Parallel Reads

For large tables, read in parallel via Spark partitioning on a numeric column:

jdbc_conf = glueContext.extract_jdbc_conf(args['connection_name'])

source_df = spark.read.format("jdbc").options(
    url=jdbc_conf["url"],
    user=jdbc_conf["user"],
    password=jdbc_conf["password"],
    dbtable="<SCHEMA>.<TABLE>",
    numPartitions=10,
    partitionColumn="<numeric_column>",
    lowerBound=1,
    upperBound="<max_value>"
).load()

Best practices:

• Use a numeric column with even distribution for partitionColumn
• Set numPartitions = number of Glue workers × 2
• Ensure lowerBound/upperBound cover actual data range
• Source database must handle concurrent connections

Retrieve credentials from the connection at runtime rather than hardcoding. See connecting-to-data-source credential-security.md for IAM DB auth and Secrets Manager patterns.

Type Mapping

Source-to-Iceberg type mappings for ingest. Apply via .cast() or column aliases in the Glue script.

Oracle

SQL Server

PostgreSQL

MySQL

Redshift

Same as PostgreSQL mappings. Redshift-specific additions:

• SUPER -> STRING (serialize) or STRUCT (parse)
• GEOMETRY / GEOGRAPHY -> BINARY or STRING

Connection Errors

If the Glue job fails with a connection-related error (timeout, auth failure, driver not found, SSL handshake), delegate to connecting-to-data-source for troubleshooting. Do not attempt network or credential fixes in this skill.

See connecting-to-data-source troubleshooting.md.

Local File Upload

Upload files from the local filesystem to S3, with optional ingestion into a table.

Workflow

1. Determine Intent

First, check the source path. If the user provides an S3 URI (e.g., s3://...) as the source, stop and use s3-files.md instead. This workflow is for local files only.

Parse the user's request to route:

• Upload only? ("put this in S3", "upload my file", "move to AWS") -> Path A
• Upload + make queryable? ("load this into a table", "ingest this CSV", "make it queryable") -> Path B

If ambiguous and the file is structured (CSV, JSON, Parquet, TSV, Avro, ORC), ask: "Do you want this queryable as a table, or just stored in S3?"

2. Discover Local Data

1. Validate path: Confirm the file or directory exists and is readable 2. Detect format: Infer from extension (.csv, .json, .parquet, .tsv, .avro, .orc) or ask 3. Check size: ls -lh for files, du -sh for directories 4. For structured files, peek at content:

• CSV/TSV: head -5 to check headers, delimiter, encoding
• JSON: head -20 to check structure (records vs. arrays)
• Parquet/Avro/ORC: note format, skip content peek

Encoding check (CSV/TSV/JSON only):

file --mime-encoding <path>

If not UTF-8 or ASCII, warn the user before upload. Non-UTF-8 files can cause downstream parsing failures.

3. Choose S3 Destination

1. Ask for target bucket or list available buckets:

   aws s3 ls

2. Suggest prefix structure: s3://<bucket>/<domain>/<dataset>/<filename> 3. Confirm with user before uploading

Default: preserve original filename. Override: user specifies a different key.

4. Upload

Single file -- check for existing objects before uploading (aws s3 cp silently overwrites):

aws s3 ls s3://<bucket>/<prefix>/<filename>

If the object exists, warn the user and get explicit confirmation before proceeding.

Directory -- check for existing objects before syncing. Use a bounded existence check to avoid enumerating every object under the prefix (which can be very slow on large prefixes):

aws s3api list-objects-v2 --bucket <bucket> --prefix <prefix>/ --max-items 1

If the result contains any Contents, objects exist and the user should be warned before proceeding. aws s3 sync skips unchanged files but overwrites modified ones without prompting.

Single file upload:

aws s3 cp <local-path> s3://<bucket>/<prefix>/<filename>

Directory upload:

aws s3 sync <local-dir> s3://<bucket>/<prefix>/

For files over 8 MB, aws s3 cp uses multipart upload automatically. No special flags needed.

Verify upload:

aws s3 ls s3://<bucket>/<prefix>/<filename>

5. Route Based on Intent

Path A: Upload Only

Report results and stop:

• S3 URI of uploaded file(s)
• File size and format
• Example command to download: aws s3 cp s3://... .

Path B: Upload + Table Ingestion

After upload completes, continue with the s3-files.md workflow using:

• S3 path where data was uploaded
• Detected file format
• Row/size estimate
• Encoding (if checked)

Do not reimplement schema inference or table creation -- follow the S3 files workflow for those steps.

Gotchas

• aws s3 cp silently overwrites existing S3 objects. Always check first.
• aws s3 sync skips unchanged files but overwrites modified ones without prompting. Check destination before syncing directories.
• CSV files with mixed encodings (e.g., Latin-1 headers, UTF-8 body) upload fine but break downstream parsing. Always check encoding for text formats.
• Large uploads on slow connections can time out. For files over 5 GB, suggest running the upload in a screen or tmux session.
• Compressed files (.gz, .zip): upload as-is for Path A. For Path B, note the compression so the S3 files workflow can handle decompression.

Troubleshooting

References

• upload-options.md -- Compression, multipart thresholds, sync vs cp tradeoffs

Upload Options Reference

cp vs sync

sync is idempotent — safe to re-run after interruption. Prefer sync for directories.

Multipart Upload

aws s3 cp automatically uses multipart for files over 8 MB (default threshold). No flags needed. To tune:

aws configure set default.s3.multipart_threshold 64MB
aws configure set default.s3.multipart_chunksize 64MB

Compression Before Upload

Compressing locally saves transfer time and storage cost. Downstream tools (Athena, Glue) read gzip natively.

gzip file.csv
aws s3 cp file.csv.gz s3://<bucket>/<prefix>/

Do NOT compress Parquet, Avro, or ORC — they have built-in compression.

Overwrite Protection

Check if target exists before uploading:

aws s3 ls s3://<bucket>/<prefix>/<filename>

If it exists, warn the user. aws s3 cp overwrites without confirmation.