AWS ETL Pipeline Patterns

Master Extract, Transform, and Load patterns with AWS Glue, Step Functions orchestration, data quality validation, and production-grade pipeline architectures.

Module: AWS Data Engineering

Topic 18 of 65

Premium Content

What is ETL?

ETL (Extract, Transform, Load) is a data integration process that moves data from source systems into a target data warehouse or data lake. It is the backbone of modern data engineering, enabling organizations to consolidate data from disparate sources for analytics and reporting.

🎯

Interview Pro Tip: This concept is frequently asked in data engineering interviews. Be ready to explain the "why" behind it, not just the "what." Connect it to real-world scenarios and trade-offs.

Core ETL Concepts

Extract: Pulling raw data from various sources (databases, APIs, files, streams)
Transform: Applying business rules, cleaning, enriching, and reshaping data
Load: Writing the processed data into a target system (data warehouse, data lake, or analytics store)

Why ETL Matters

Aspect	Impact
Data Quality	Enforces validation rules before data reaches consumers
Consistency	Standardizes data formats across heterogeneous sources
Performance	Pre-aggregates and indexes data for faster queries
Compliance	Applies masking and anonymization for GDPR/HIPAA
Cost Control	Filters unnecessary data before it enters expensive storage

ETL vs ELT

Feature	ETL	ELT
Transform Location	External processing engine	Inside the target warehouse
Best For	Complex transformations, compliance	Large datasets, cloud-native warehouses
Latency	Higher (extra processing step)	Lower (load raw, transform in-place)
AWS Services	Glue, EMR, Lambda	Redshift, Athena, Spark on EMR
Storage Cost	Lower (only clean data stored)	Higher (raw + transformed stored)

📝

Deep Dive: ETL vs ELT

Modern data engineering has shifted from ETL (transform before load) to ELT (load then transform). The ELT approach leverages the compute power of modern data warehouses like Redshift and Snowflake. Learn the trade-offs in our ETL vs ELT guide and how dbt enables SQL-first transformations.

Extract Patterns

The extraction phase pulls raw data from multiple heterogeneous sources into a staging area. AWS provides connectors and services for virtually every data source.

Pattern 1: S3 Batch Extraction

import boto3
from awsglue.transforms import *
from awsglue.context import GlueContext
from pyspark.context import SparkContext

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

# Extract from S3 with partition pruning
dynamic_frame = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={
        "paths": ["s3://data-lake/raw/sales/"],
        "groupBy": ["year", "month"],
        "recurse": True,
        "enablePartitionDiscovery": True
    },
    format="parquet",
    transformation_ctx="s3_extract"
)

print(f"Extracted {dynamic_frame.count()} records")

Pattern 2: RDS Incremental Extraction (CDC)

import boto3

# Use Glue CDC for PostgreSQL
cdc_options = {
    "connectionName": "rds-postgres-conn",
    "table": "public.orders",
    "database": "production",
    "cdcOptions": {
        "capturedInstance": "orders-table",
        "readFromWriteTimestamp": True
    },
    "outputFormat": "dynamicframe",
    "startingPosition": "last-sync"
}

cdc_frame = glueContext.create_dynamic_frame.from_jdbc_conf(
    catalog_connection=cdc_options["connectionName"],
    database=cdc_options["database"],
    table=cdc_options["table"],
    transformation_ctx="cdc_extract"
)

Pattern 3: API Extraction with Pagination

import requests
import json
from datetime import datetime

class APIDataExtractor:
    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self.session = requests.Session()

    def extract_paginated(self, endpoint, page_size=1000):
        """Extract all data with cursor-based pagination."""
        all_records = []
        cursor = None
        page = 0

        while True:
            params = {"limit": page_size}
            if cursor:
                params["cursor"] = cursor

            response = self.session.get(
                f"{self.base_url}/{endpoint}",
                headers=self.headers,
                params=params
            )
            response.raise_for_status()
            data = response.json()

            records = data.get("results", [])
            all_records.extend(records)
            page += 1

            cursor = data.get("next_cursor")
            if not cursor or len(records) == 0:
                break

        return all_records

    def extract_to_s3(self, endpoint, bucket, key_prefix):
        """Extract API data and save to S3."""
        records = self.extract_paginated(endpoint)
        timestamp = datetime.utcnow().isoformat()

        s3_client = boto3.client("s3")
        s3_client.put_object(
            Bucket=bucket,
            Key=f"{key_prefix}/{timestamp}.json",
            Body=json.dumps(records)
        )
        return len(records)

Pattern 4: Kinesis Stream Extraction

import boto3
import json

kinesis_client = boto3.client("kinesis")

def extract_from_kinesis(shard_id, iterator_type="LATEST"):
    """Extract records from a Kinesis shard."""
    response = kinesis_client.get_shard_iterator(
        StreamName="real-time-events",
        ShardId=shard_id,
        ShardIteratorType=iterator_type
    )

    shard_iterator = response["ShardIterator"]
    records_batch = []

    while shard_iterator:
        response = kinesis_client.get_records(
            ShardIterator=shard_iterator,
            Limit=1000
        )
        records_batch.extend(response["Records"])
        shard_iterator = response.get("NextShardIterator")

        if len(records_batch) >= 10000:
            yield records_batch
            records_batch = []

    if records_batch:
        yield records_batch

Transform Patterns

Transformations clean, enrich, reshape, and aggregate raw data into business-ready formats. AWS Glue provides a rich set of built-in transforms plus Spark SQL for custom logic.

Pattern 1: Data Cleaning and Validation

from awsglue.transforms import *
from pyspark.sql import functions as F
from pyspark.sql.types import *

def clean_data(dynamic_frame):
    """Apply comprehensive data cleaning."""
    df = dynamic_frame.toDF()

    # Remove duplicates based on business key
    df = df.dropDuplicates(["order_id", "customer_id"])

    # Handle null values
    df = df.fillna({
        "amount": 0.0,
        "status": "UNKNOWN",
        "region": "UNASSIGNED"
    })

    # Validate and filter invalid records
    df = df.filter(
        (F.col("amount") > 0) &
        (F.col("order_date").isNotNull()) &
        (F.col("customer_id").rlike("^CUST-[0-9]{6}$"))
    )

    # Standardize formats
    df = df.withColumn(
        "order_date",
        F.to_date(F.col("order_date"), "yyyy-MM-dd")
    )
    df = df.withColumn(
        "email",
        F.lower(F.trim(F.col("email")))
    )

    return df

Pattern 2: Data Enrichment

def enrich_orders(orders_df, customers_df, products_df):
    """Enrich orders with customer and product details."""
    # Join with customer dimension
    enriched = orders_df.join(
        customers_df,
        orders_df.customer_id == customers_df.customer_id,
        "left"
    ).select(
        orders_df["*"],
        customers_df["customer_name"],
        customers_df["segment"],
        customers_df["country"]
    )

    # Join with product dimension
    enriched = enriched.join(
        products_df,
        orders_df.product_id == products_df.product_id,
        "left"
    ).select(
        enriched["*"],
        products_df["product_name"],
        products_df["category"],
        products_df["unit_cost"]
    )

    # Add computed columns
    enriched = enriched.withColumn(
        "profit_margin",
        (F.col("amount") - F.col("unit_cost")) / F.col("amount")
    )
    enriched = enriched.withColumn(
        "order_tier",
        F.when(F.col("amount") >= 1000, "Enterprise")
         .when(F.col("amount") >= 100, "Mid-Market")
         .otherwise("SMB")
    )

    return enriched

Pattern 3: Aggregation and Pivoting

def aggregate_sales(enriched_df):
    """Create aggregated sales summaries."""
    # Daily aggregation
    daily_sales = enriched_df.groupBy(
        F.col("order_date"),
        F.col("category"),
        F.col("region")
    ).agg(
        F.count("order_id").alias("order_count"),
        F.sum("amount").alias("total_revenue"),
        F.avg("amount").alias("avg_order_value"),
        F.countDistinct("customer_id").alias("unique_customers")
    )

    # Monthly pivot by category
    monthly_pivot = daily_sales.groupBy(
        F.month("order_date").alias("month"),
        F.col("region")
    ).pivot("category", ["Electronics", "Software", "Services"])
    .agg(
        F.sum("total_revenue").alias("revenue")
    )

    return daily_sales, monthly_pivot

Pattern 4: Data Masking for Compliance

import hashlib

def mask_pii(dynamic_frame):
    """Apply PII masking for GDPR/CCPA compliance."""
    df = dynamic_frame.toDF()

    # Hash sensitive columns
    df = df.withColumn(
        "email_masked",
        F.sha2(F.col("email"), 256)
    )
    df = df.withColumn(
        "phone_masked",
        F.concat(
            F.lit("***-***-"),
            F.substring(F.col("phone"), -4, 4)
        )
    )

    # Redact full name
    df = df.withColumn(
        "customer_name",
        F.concat(
            F.substring(F.col("customer_name"), 1, 1),
            F.lit("***")
        )
    )

    # Drop original PII columns
    df = df.drop("email", "phone", "ssn")

    return df

Load Patterns

The load phase writes transformed data to target systems optimized for analytics, reporting, and application consumption.

Pattern 1: Redshift Bulk Load

def load_to_redshift(transformed_df, table_name):
    """Load data into Redshift using COPY command."""
    transformed_df.write \
        .format("com.databricks.spark.redshift") \
        .option("url", "jdbc:redshift://cluster.xxxx.us-east-1.redshift.amazonaws.com:5439/dev") \
        .option("dbtable", table_name) \
        .option("tempdir", "s3://redshift-temp/loading/") \
        .option("aws_iam_role", "arn:aws:iam::role/RedshiftLoadRole") \
        .option("diststyle", "KEY") \
        .option("distkey", "customer_id") \
        .option("sortkeys", "order_date") \
        .mode("append") \
        .save()

Pattern 2: S3 Data Lake Write (Partitioned)

def load_to_datalake(transformed_df, base_path, partition_cols):
    """Write to S3 data lake with optimized partitioning."""
    transformed_df.write \
        .mode("overwrite") \
        .partitionBy(*partition_cols) \
        .option("compression", "snappy") \
        .parquet(f"{base_path}/processed/")

    # Register in Glue Catalog
    glueContext.write_dynamic_frame.from_options(
        frame=dynamic_frame,
        connection_type="s3",
        connection_options={"path": f"{base_path}/processed/"},
        format="parquet",
        catalog_info={
            "database": "analytics_db",
            "table_name": table_name
        }
    )

Pattern 3: DynamoDB Write with Batch Operations

import boto3

def load_to_dynamodb(records, table_name):
    """Batch write records to DynamoDB."""
    dynamodb = boto3.resource("dynamodb")
    table = dynamodb.Table(table_name)

    with table.batch_writer(
        overwrite_by_pkeys=["PK", "SK"]
    ) as batch:
        for record in records:
            batch.put_item(Item={
                "PK": f"ORDER#{record['order_id'
  ]}",
                "SK": f"CUST#{record['customer_id'
  ]}",
                "amount": record["amount"],
                "status": record["status"],
                "ttl": record.get("ttl", 0)
            })

Pattern 4: Athena Query Results to S3

import boto3

athena_client = boto3.client("athena")

def load_via_athena(query, output_bucket):
    """Execute query and persist results to S3."""
    response = athena_client.start_query_execution(
        QueryString=query,
        QueryExecutionContext={"Database": "analytics_db"},
        ResultConfiguration={
            "OutputLocation": f"s3://{output_bucket}/athena-results/"
        }
    )
    return response["QueryExecutionId"]

Glue ETL Pipeline Architecture

A production-grade Glue ETL pipeline combines multiple AWS services for orchestration, processing, monitoring, and delivery.

Architecture Components

📝

Key Concept: Understanding this architecture is essential for designing scalable, cost-effective data platforms on AWS. Draw this diagram from memory during interviews.

Component	Service	Purpose
Orchestration	Step Functions	Coordinates pipeline stages with error handling
Extraction	Glue Crawlers	Discovers schemas and populates Data Catalog
Processing	Glue ETL Jobs	Spark-based data transformation
Validation	Lambda	Runs custom validation logic
Loading	Glue Connectors	Writes to Redshift, RDS, and other targets
Monitoring	CloudWatch	Logs, metrics, and alarms
Alerting	SNS	Sends notifications on failures
Storage	S3	Data lake storage for all stages

Full Pipeline Implementation

# glue_etl_job.py - Main transformation job
import sys
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.context import SparkContext
from pyspark.sql import functions as F

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

# Read from raw zone
raw_orders = glueContext.create_dynamic_frame.from_catalog(
    database="raw_db",
    table_name="orders",
    transformation_ctx="raw_orders"
)

raw_customers = glueContext.create_dynamic_frame.from_catalog(
    database="raw_db",
    table_name="customers",
    transformation_ctx="raw_customers"
)

# Transform
orders_df = raw_orders.toDF()
customers_df = raw_customers.toDF()

# Clean
orders_clean = orders_df \
    .dropDuplicates(["order_id"]) \
    .filter(F.col("amount") > 0) \
    .fillna({"status": "UNKNOWN"})

# Enrich
enriched = orders_clean.join(
    customers_df,
    orders_clean.customer_id == customers_df.customer_id,
    "left"
).select(
    orders_clean["*"],
    customers_df["customer_name"],
    customers_df["segment"]
)

# Write to processed zone
result = glueContext.create_dynamic_frame.from_options(
    frame=glueContext.create_dynamic_frame.fromDF(
        enriched, glueContext, "enriched"
    ),
    connection_type="s3",
    connection_options={
        "path": "s3://data-lake/processed/orders/",
        "partitionKeys": ["year", "month"]
    },
    format="parquet",
    format_options={"compression": "snappy"},
    transformation_ctx="processed_output"
)

job.commit()

Step Functions State Machine Definition

{
  "Comment": "Production ETL Pipeline with Error Handling",
  "StartAt": "RunCrawler",
  "States": {
    "RunCrawler": {
      "Type": "Task",
      "Resource": "arn:aws:states:::glue:startCrawler.sync",
      "Parameters": {
        "Name": "raw-data-crawler"
      },
      "Next": "ExtractData",
      "Retry": [
        {
          "ErrorEquals": ["Glue.CrawlerRunningException"],
          "IntervalSeconds": 30,
          "MaxAttempts": 3,
          "BackoffRate": 2
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "HandleError",
          "ResultPath": "$.error"
        }
      ]
    },
    "ExtractData": {
      "Type": "Task",
      "Resource": "arn:aws:states:::glue:startJobRun.sync",
      "Parameters": {
        "JobName": "etl-extract-job",
        "Arguments": {
          "--source": "s3://raw-zone/",
          "--date": "$.date"
        }
      },
      "Next": "ValidateData",
      "Retry": [
        {
          "ErrorEquals": ["Glue.EntityNotFoundException", "Glue.ConcurrentRunsException"],
          "IntervalSeconds": 60,
          "MaxAttempts": 2,
          "BackoffRate": 2
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "HandleError",
          "ResultPath": "$.error"
        }
      ]
    },
    "ValidateData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:function:etl-validate-data",
      "InputPath": "$.JobRun.Output",
      "Next": "CheckQuality",
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "HandleError",
          "ResultPath": "$.error"
        }
      ]
    },
    "CheckQuality": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.quality_score",
          "NumericGreaterThanEquals": 95,
          "Next": "TransformData"
        }
      ],
      "Default": "QuarantineData"
    },
    "TransformData": {
      "Type": "Task",
      "Resource": "arn:aws:states:::glue:startJobRun.sync",
      "Parameters": {
        "JobName": "etl-transform-job",
        "Arguments": {
          "--input": "$.staging_path",
          "--output": "$.processed_path"
        }
      },
      "Next": "LoadData",
      "Retry": [
        {
          "ErrorEquals": ["States.TaskFailed"],
          "IntervalSeconds": 120,
          "MaxAttempts": 2
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "HandleError",
          "ResultPath": "$.error"
        }
      ]
    },
    "LoadData": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "LoadRedshift",
          "States": {
            "LoadRedshift": {
              "Type": "Task",
              "Resource": "arn:aws:states:::glue:startJobRun.sync",
              "Parameters": {
                "JobName": "etl-load-redshift"
              },
              "End": true
            }
          }
       }, {
          "StartAt": "LoadDataLake",
          "States": {
            "LoadDataLake": {
              "Type": "Task",
              "Resource": "arn:aws:states:::glue:startJobRun.sync",
              "Parameters": {
                "JobName": "etl-load-s3-datalake"
              },
              "End": true
            }
          }
        }
      ],
      "Next": "SendSuccessNotification",
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "HandleError",
          "ResultPath": "$.error"
        }
      ]
    },
    "SendSuccessNotification": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:function:etl-notify-success",
      "End": true
    },
    "QuarantineData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:function:etl-quarantine",
      "Next": "SendQualityAlert",
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "HandleError"
        }
      ]
    },
    "SendQualityAlert": {
      "Type": "Task",
      "Resource": "arn:aws:sns:us-east-1:topic/etl-quality-alert",
      "Parameters": {
        "Message": "Data quality below threshold. Records quarantined.",
        "Subject": "ETL Quality Alert"
      },
      "End": true
    },
    "HandleError": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:function:etl-handle-error",
      "Parameters": {
        "error.$": "$.error",
        "pipeline": "daily-etl",
        "timestamp.$": "&dollar;&dollar;.State.EnteredTime"
      },
      "Next": "SendErrorAlert"
    },
    "SendErrorAlert": {
      "Type": "Task",
      "Resource": "arn:aws:sns:us-east-1:topic/etl-pipeline-alerts",
      "Parameters": {
        "Message": "ETL pipeline failed. Check CloudWatch logs.",
        "Subject": "ETL Pipeline Failure"
      },
      "End": true
    }
  }
}

Error Handling and Monitoring

Production ETL pipelines require robust error handling, dead-letter queues, and comprehensive monitoring to ensure data reliability.

Error Handling Strategies

Strategy	Use Case	AWS Service
Retry with Backoff	Transient failures (throttling, network)	Step Functions Retry
Dead Letter Queue	Permanently failed records	SQS / SNS
Circuit Breaker	Downstream service overload	Lambda + CloudWatch
Idempotent Retries	Duplicate processing prevention	DynamoDB conditions
Data Quarantine	Invalid records needing manual review	S3 + Lambda
Alert Escalation	Critical failures requiring human intervention	SNS + PagerDuty

Error Handler Lambda

import boto3
import json
from datetime import datetime

sns_client = boto3.client("sns")
dynamodb = boto3.resource("dynamodb")
error_table = dynamodb.Table("etl-error-log")

def lambda_handler(event, context):
    """Central error handler for ETL pipeline."""
    error = event.get("error", {})
    pipeline = event.get("pipeline", "unknown")
    timestamp = event.get("timestamp", datetime.utcnow().isoformat())

    # Log error to DynamoDB
    error_table.put_item(Item={
        "pipeline": pipeline,
        "timestamp": timestamp,
        "error_type": error.get("Error", "Unknown"),
        "error_message": error.get("Cause", "No details"),
        "state_name": error.get("StateName", "Unknown"),
        "ttl": int((datetime.utcnow().timestamp()) + (90 * 24 * 3600))
    })

    # Classify error severity
    severity = classify_error(error)

    # Send appropriate notification
    if severity == "CRITICAL":
        sns_client.publish(
            TopicArn="arn:aws:sns:us-east-1:topic/etl-critical-alerts",
            Subject=f"CRITICAL: {pipeline} pipeline failure",
            Message=json.dumps({
                "pipeline": pipeline,
                "error": error,
                "action_required": True,
                "timestamp": timestamp
            }, indent=2)
        )
    elif severity == "WARNING":
        sns_client.publish(
            TopicArn="arn:aws:sns:us-east-1:topic/etl-warning-alerts",
            Subject=f"WARNING: {pipeline} pipeline issue",
            Message=json.dumps({
                "pipeline": pipeline,
                "error": error,
                "action_required": False
            }, indent=2)
        )

    return {"severity": severity, "logged": True}

def classify_error(error):
    """Classify error severity based on type."""
    error_type = error.get("Error", "")

    critical_patterns = [
        "AccessDeniedException",
        "KMSUnauthorizedAccess",
        "DataCatalogNotFound"
    ]
    warning_patterns = [
        "ThrottlingException",
        "TooManyRequestsException",
        "ServiceUnavailable"
    ]

    for pattern in critical_patterns:
        if pattern in error_type:
            return "CRITICAL"

    for pattern in warning_patterns:
        if pattern in error_type:
            return "WARNING"

    return "ERROR"

CloudWatch Dashboard Metrics

import boto3

cloudwatch = boto3.client("cloudwatch")

def create_etl_dashboard(pipeline_name):
    """Create CloudWatch dashboard for ETL monitoring."""
    dashboard_body = {
        "widgets": [
            {
                "type": "metric",
                "x": 0, "y": 0,
                "width": 12, "height": 6,
                "properties": {
                    "metrics": [
                        ["AWS/Glue", "glue.${JobName}.memoryUsed", "JobName", job]
                        for job in ["extract", "transform", "load"]
                    ],
                    "period": 300,
                    "stat": "Average",
                    "title": "Glue Job Memory Usage"
                }
           }, {
                "type": "metric",
                "x": 12, "y": 0,
                "width": 12, "height": 6,
                "properties": {
                    "metrics": [
                        ["AWS/Glue", "glue.${JobName}.jobDuration", "JobName", job]
                        for job in ["extract", "transform", "load"]
                    ],
                    "period": 300,
                    "stat": "Sum",
                    "title": "Glue Job Duration (seconds)"
                }
           }, {
                "type": "metric",
                "x": 0, "y": 6,
                "width": 12, "height": 6,
                "properties": {
                    "metrics": [
                        ["AWS/StepFunctions", "ExecutionsFailed", "StateMachineArn", "etl-pipeline"],
                        ["AWS/StepFunctions", "ExecutionsSucceeded", "StateMachineArn", "etl-pipeline"]
                    ],
                    "period": 3600,
                    "stat": "Sum",
                    "title": "Pipeline Execution Status"
                }
           }, {
                "type": "metric",
                "x": 12, "y": 6,
                "width": 12, "height": 6,
                "properties": {
                    "metrics": [
                        ["AWS/SQS", "ApproximateNumberOfMessagesVisible", "QueueName", "etl-dead-letter-queue"]
                    ],
                    "period": 300,
                    "stat": "Average",
                    "title": "Dead Letter Queue Depth"
                }
            }
        ]
    }

    cloudwatch.put_dashboard(
        DashboardName=f"ETL-Pipeline-{pipeline_name}",
        DashboardBody=json.dumps(dashboard_body)
    )

Architecture Flow

Interview Q&A

Q1: When should you use AWS Glue vs. AWS Lambda for ETL?

Answer: Use Glue for large datasets (>1GB), complex multi-step transformations requiring Spark, and when you need built-in data catalog integration. Use Lambda for small datasets (<6GB payload), simple transformations, event-driven triggers, and when sub-second startup latency matters. Glue provides auto-scaling of workers, while Lambda has a 15-minute execution limit and 6GB memory ceiling.

Q2: How do you implement idempotent ETL pipelines?

Answer: Idempotency ensures that re-running a pipeline produces the same result without data duplication. Key techniques include:

Job Bookmarks: Glue tracks processed data automatically
Partition Overwriting: Write to date-based partitions with mode("overwrite")
DynamoDB Dedup Table: Record processed batch IDs with conditional writes
Atomic S3 Writes: Use unique output paths with PUT operations
Watermarking: Track high-water marks in a metadata store

Q3: What is the advantage of using Step Functions over simple Lambda chains?

Answer: Step Functions provides visual workflow orchestration, built-in error handling with retry/catch, state tracking across executions, support for human approval steps, execution history for debugging, and the ability to orchestrate Glue jobs directly via .sync integrations. Lambda chains require manual error handling, state management, and provide no built-in observability.

Q4: How do you handle schema evolution in Glue ETL pipelines?

Answer: Use Glue's resolveChoice method to handle schema mismatches, enable pushdownPredicate for partition evolution, and implement schema versioning in the Data Catalog. For breaking changes, use a two-phase approach: first deploy a schema migration job, then update downstream consumers. Glue Studio visual jobs make schema drift detection easier through its data quality panel.

Q5: How do you optimize Glue ETL job performance?

Answer: Key optimization strategies include:

Right-sizing workers: Start with G.1X (4 vCPU, 16GB), scale up for memory-heavy transforms
Partition pruning: Filter early and use pushdownPredicate
Columnar formats: Read/write Parquet or ORC instead of JSON/CSV
Dynamic frames vs DataFrames: Use DataFrames for complex joins, dynamic frames for schema flexibility
Job bookmarking: Enable incremental processing to avoid full scans
Auto scaling: Use Glue's auto-scaling workers for variable workloads

Q6: Explain the difference between Glue crawlers and Glue ETL jobs.

Answer: Crawlers automatically discover schema, infer data types, and populate the Glue Data Catalog with table definitions and partitions. They are read-only discovery tools. ETL Jobs are compute processes that read, transform, and write data using Spark or Python Shell. Crawlers should run before ETL jobs to ensure the catalog is current, while ETL jobs consume catalog metadata to locate and process data.

Q7: How do you monitor data quality in production ETL pipelines?

Answer: Implement a multi-layered monitoring approach:

Pre-stage validation: Lambda functions validate schema and nulls before transformation
In-job quality checks: Spark assertions with assert statements
Post-stage profiling: Glue DataBrew for statistical profiling
CloudWatch metrics: Custom metrics for record counts, null rates, and freshness
Alerting: SNS topics for quality threshold breaches
Quarantine: Route failed records to S3 quarantine bucket for manual review

Q8: What are the best practices for Glue job security?

Answer:

IAM roles: Least-privilege roles for each job type
Encryption: Enable at-rest encryption for S3 targets and in-transit for JDBC connections
Secrets Manager: Store database credentials centrally, never hardcode
VPC endpoints: Keep traffic within AWS network for RDS sources
CloudTrail: Audit all Glue API calls
Tags: Tag all resources for cost allocation and access control

Q9: How do you design a Glue ETL pipeline for near-real-time data?

Answer: Use Glue Streaming ETL for continuous processing from Kinesis or Kafka. Configure micro-batch intervals (1-5 minutes), use windowed aggregations for time-based summaries, and implement watermarks for late-arriving data. For sub-minute latency, combine Kinesis Data Analytics with Lambda triggers. Glue Streaming supports exactly-once processing with checkpointing.

Q10: How do you handle failures and retries in a multi-step Glue pipeline?

Answer: Design with a Step Functions state machine that wraps each Glue job. Configure retry policies with exponential backoff for transient errors (throttling, network). Use Catch blocks to route to error handlers. Implement a dead letter queue for records that fail repeatedly. Use idempotent writes so retries are safe. Store pipeline state in DynamoDB to enable resume-from-failure capabilities.

Summary

ETL (Extract, Transform, Load) is the foundation of data warehousing and analytics pipelines
AWS Glue provides serverless Spark-based ETL with auto-scaling, data catalog, and job bookmarks
Step Functions orchestrates multi-step pipelines with visual workflows, error handling, and retry logic
Transform patterns include cleaning, enrichment, aggregation, pivoting, and PII masking
Load patterns span Redshift, S3 data lake, DynamoDB, and OpenSearch targets
Error handling requires classification, dead letter queues, quarantine, and tiered alerting
Monitoring combines CloudWatch metrics, custom dashboards, and SNS notifications
Idempotency is critical for reliable retries and exactly-once processing guarantees
Schema evolution must be planned with catalog versioning and migration strategies
Security follows least-privilege IAM, encryption, and secrets management best practices

AWS ETL Pipeline Patterns for Data Engineers