AWS Batch Analytics for Data Engineers

Master batch analytics on AWS with EMR, Glue, Athena, and large-scale batch processing patterns for production data pipelines.

25 min readAdvanced

Why This Matters

Batch analytics is the backbone of enterprise data processing on AWS. It involves collecting, processing, and analyzing large volumes of data in scheduled or triggered batches rather than in real-time. AWS provides a comprehensive suite of services that work together to build scalable, cost-effective batch analytics solutions. Understanding batch patterns is critical for data engineers because most business intelligence, reporting, and machine learning training pipelines operate in batch mode. Mastering these patterns directly impacts your ability to design cost-effective, reliable data platforms that serve downstream consumers including analysts, data scientists, and business applications.

Batch Analytics Architecture

Real-World Project Structure

Architecture Diagram

aws-batch-analytics/
├── infrastructure/
│   ├── terraform/
│   │   ├── main.tf                    # EMR cluster, Glue crawlers, S3 buckets
│   │   ├── variables.tf               # Environment, instance types, scaling
│   │   └── outputs.tf                 # Cluster IDs, endpoint URLs
│   └── cloudformation/
│       └── batch-stack.yaml           # Step Functions, IAM roles, alarms
├── pipelines/
│   ├── emr/
│   │   ├── jobs/
│   │   │   ├── silver_transform.py    # Bronze to Silver transformation
│   │   │   ├── gold_aggregate.py      # Silver to Gold aggregation
│   │   │   └── data_quality.py        # Validation and cleansing
│   │   └── configs/
│   │       └── spark-defaults.conf    # Spark tuning parameters
│   ├── glue/
│   │   ├── crawlers/
│   │   │   └── s3_crawler.json        # Crawler definitions
│   │   └── jobs/
│   │       └── etl_job.py             # Glue ETL script with bookmarks
│   └── athena/
│       ├── queries/
│       │   ├── daily_summary.sql      # Standard queries
│       │   └── ad_hoc.sql             # Ad-hoc analysis
│       └── workgroups/
│           └── analytics.json         # Workgroup configuration
├── orchestration/
│   ├── step_functions/
│   │   └── batch_workflow.asl.json    # State machine definition
│   └── scheduling/
│       └── eventbridge_rules.json     # Cron triggers
├── monitoring/
│   ├── cloudwatch/
│   │   ├── alarms.json                # Job failure alarms
│   │   └── dashboards.json            # Operational dashboards
│   └── logging/
│       └── log_groups.json            # Log retention policies
├── tests/
│   ├── unit/
│   │   ├── test_transform.py          # Transformation unit tests
│   │   └── test_quality.py            # Data quality tests
│   └── integration/
│       └── test_pipeline.py           # End-to-end pipeline tests
└── scripts/
    ├── deploy.sh                      # CI/CD deployment script
    ├── validate.sh                    # Pre-deployment validation
    └── monitoring.sh                  # Health check script

EMR for Batch Analytics

Amazon EMR is the primary service for running large-scale batch analytics workloads on AWS. It provides managed clusters running open-source frameworks like Apache Spark, Hive, and Presto.

EMR Instance Selection for Batch Workloads

Workload Type	Recommended Instances	Use Case
Memory-intensive	r5/r6g, r5d	Spark SQL, joins, aggregations
Compute-intensive	c5/c6g, c5d	MapReduce, machine learning
Storage-intensive	i3, d2	Large dataset processing
Cost-optimized	m5/m6g, Spot	Flexible batch jobs

EMR Batch Job Production Code

"""
EMR Batch Processing Job with error handling and checkpointing.
Processes Bronze layer data to Silver layer with validation.
"""
import sys
import logging
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, current_timestamp, input_file_name, lit, 
    when, count, sum as spark_sum
)
from pyspark.sql.types import StructType, StructField, StringType, LongType, DecimalType

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def create_spark_session():
    """Create optimized Spark session for batch processing."""
    return SparkSession.builder \
        .appName("BatchAnalytics_Silver") \
        .config("spark.sql.shuffle.partitions", "200") \
        .config("spark.sql.files.maxPartitionBytes", "134217728") \
        .config("spark.sql.adaptive.enabled", "true") \
        .config("spark.dynamicAllocation.enabled", "true") \
        .config("spark.dynamicAllocation.minExecutors", "2") \
        .config("spark.dynamicAllocation.maxExecutors", "50") \
        .config("spark.sql.sources.partitionOverwriteMode", "dynamic") \
        .getOrCreate()

def validate_schema(df, expected_schema):
    """Validate DataFrame schema matches expected structure."""
    errors = []
    for field in expected_schema:
        if field.name not in df.columns:
            errors.append(f"Missing column: {field.name}")
        elif str(df.schema[field.name].dataType) != str(field.dataType):
            errors.append(f"Type mismatch: {field.name} expected {field.dataType}")
    if errors:
        raise ValueError(f"Schema validation failed: {'; '.join(errors)}")
    logger.info(f"Schema validation passed for {len(expected_schema)} columns")
    return True

def process_bronze_to_silver(spark, input_path, output_path, batch_id):
    """Transform Bronze data to Silver layer with quality checks."""
    try:
        logger.info(f"Starting batch processing: {batch_id}")
        
        # Read with schema validation
        raw_df = spark.read.json(input_path)
        record_count = raw_df.count()
        logger.info(f"Read {record_count} records from {input_path}")
        
        if record_count == 0:
            logger.warning("No records found. Skipping transformation.")
            return
        
        # Add metadata columns
        enriched_df = raw_df \
            .withColumn("ingestion_timestamp", current_timestamp()) \
            .withColumn("source_file", input_file_name()) \
            .withColumn("batch_id", lit(batch_id))
        
        # Deduplicate by primary key
        window_spec = Window.partitionBy("event_id").orderBy(col("ingestion_timestamp").desc())
        deduped_df = enriched_df \
            .withColumn("row_num", row_number().over(window_spec)) \
            .filter(col("row_num") == 1) \
            .drop("row_num")
        
        # Apply data quality rules
        valid_df = deduped_df.filter(
            col("event_id").isNotNull() &
            col("user_id").isNotNull() &
            (col("amount") >= 0) &
            (col("amount") <= 1000000)
        )
        
        invalid_count = deduped_df.count() - valid_df.count()
        if invalid_count > 0:
            logger.warning(f"Filtered {invalid_count} invalid records")
        
        # Write with partitioning
        valid_df.write \
            .mode("overwrite") \
            .partitionBy("event_date", "event_type") \
            .parquet(output_path)
        
        logger.info(f"Successfully wrote {valid_df.count()} records to {output_path}")
        
    except Exception as e:
        logger.error(f"Batch processing failed: {str(e)}")
        raise

def main():
    """Main entry point with CLI argument handling."""
    try:
        spark = create_spark_session()
        
        # Parse arguments
        input_path = sys.argv[1] if len(sys.argv) > 1 else "s3://data-lake/bronze/events/"
        output_path = sys.argv[2] if len(sys.argv) > 2 else "s3://data-lake/silver/events/"
        batch_id = sys.argv[3] if len(sys.argv) > 3 else "manual_run"
        
        process_bronze_to_silver(spark, input_path, output_path, batch_id)
        
    except Exception as e:
        logger.error(f"Job failed with error: {str(e)}")
        sys.exit(1)
    finally:
        if 'spark' in locals():
            spark.stop()

if __name__ == "__main__":
    from pyspark.sql.window import Window
    from pyspark.sql.functions import row_number
    main()

EMR Cluster Creation with Cost Optimization

#!/bin/bash
# Create EMR cluster with Spot instances for cost optimization

CLUSTER_NAME="batch-analytics-$(date +%Y%m%d)"
LOG_URI="s3://data-lake-emr-logs/emr-logs/"

aws emr create-cluster \
    --name "$CLUSTER_NAME" \
    --release-label emr-6.15.0 \
    --applications Name=Spark Name=Hive Name=JupyterEnterpriseGateway \
    --instance-fleets '[
      {
        "InstanceFleetType": "MASTER",
        "TargetOnDemandCapacity": 1,
        "InstanceTypeConfigs": [
          {
            "InstanceType": "m5.xlarge",
            "EbsConfiguration": {
              "EbsBlockDeviceConfigs": [
                {"VolumeSpecification": {"Size": 100, "Type": "gp3"}, "VolumesPerInstance": 1}
              ]
            }
          }
        ]
      },
      {
        "InstanceFleetType": "CORE",
        "TargetOnDemandCapacity": 2,
        "TargetSpotCapacity": 8,
        "InstanceTypeConfigs": [
          {
            "InstanceType": "m5.xlarge",
            "WeightedCapacity": 1,
            "EbsConfiguration": {
              "EbsBlockDeviceConfigs": [
                {"VolumeSpecification": {"Size": 200, "Type": "gp3"}, "VolumesPerInstance": 2}
              ]
            }
          },
          {
            "InstanceType": "m5.2xlarge",
            "WeightedCapacity": 2,
            "EbsConfiguration": {
              "EbsBlockDeviceConfigs": [
                {"VolumeSpecification": {"Size": 400, "Type": "gp3"}, "VolumesPerInstance": 2}
              ]
            }
          }
        ]
      }
    ]' \
    --ec2-attributes KeyName=my-key-pair,SubnetIds=subnet-1a2b3c4d,subnet-5e6f7g8h \
    --service-role EMR_DefaultRole \
    --ec2-role EMR_EC2_DefaultRole \
    --log-uri "$LOG_URI" \
    --enable-application-portals \
    --configurations '[
      {
        "Classification": "spark-defaults",
        "Properties": {
          "spark.dynamicAllocation.enabled": "true",
          "spark.dynamicAllocation.minExecutors": "2",
          "spark.dynamicAllocation.maxExecutors": "50",
          "spark.sql.adaptive.enabled": "true",
          "spark.sql.shuffle.partitions": "200"
        }
      }
    ]' \
    --tags Environment=production,Project=batch-analytics \
    --auto-terminate

AWS Glue for Batch ETL

AWS Glue provides a serverless ETL service that makes it easy to prepare and load data for analytics. It automatically discovers and catalogs data from various sources.

Glue ETL Job with Bookmarks

"""
AWS Glue ETL Job with incremental processing via bookmarks.
"""
import sys
import logging
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, when, current_timestamp

logger = logging.getLogger(__name__)

args = getResolvedOptions(sys.argv, [
    'JOB_NAME', 'SOURCE_DATABASE', 'SOURCE_TABLE',
    'TARGET_DATABASE', 'TARGET_TABLE'
])

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

try:
    # Read with bookmark for incremental processing
    datasource = glueContext.create_dynamic_frame.from_catalog(
        database=args['SOURCE_DATABASE'],
        table_name=args['SOURCE_TABLE'],
        transformation_ctx="datasource",
        additional_options={"enableUpdateCatalog": True}
    )
    
    logger.info(f"Read {datasource.count()} records with bookmark")
    
    # Apply mappings and transformations
    apply_mapping = ApplyMapping.apply(
        frame=datasource,
        mappings=[
            ("event_id", "string", "event_id", "string"),
            ("user_id", "long", "user_id", "long"),
            ("amount", "decimal", "amount", "decimal(10,2)"),
            ("event_type", "string", "event_type", "string"),
            ("event_date", "string", "event_date", "string")
        ],
        transformation_ctx="apply_mapping"
    )
    
    # Write with bookmark tracking
    sink = glueContext.write_dynamic_frame.from_catalog(
        frame=apply_mapping,
        database=args['TARGET_DATABASE'],
        table_name=args['TARGET_TABLE'],
        transformation_ctx="sink",
        enable_update_catalog=True,
        update_catalog_options={"updateBehavior": "UPDATE_IN_DATABASE"}
    )
    
    logger.info("ETL job completed successfully")
    
except Exception as e:
    logger.error(f"Glue job failed: {str(e)}")
    raise

job.commit()

Athena for Ad-hoc Analytics

Amazon Athena is an interactive query service that makes it easy to analyze data in Amazon S3 using standard SQL. It is serverless with no infrastructure to manage.

Athena Query Optimization

-- Partitioned query for cost-efficient analysis
SELECT 
    event_date,
    event_type,
    COUNT(*) as event_count,
    COUNT(DISTINCT user_id) as unique_users,
    SUM(amount) as total_amount
FROM events
WHERE year = '2026' 
  AND month = '07' 
  AND day = '15'
GROUP BY event_date, event_type
ORDER BY event_count DESC;

-- CTAS for materialized summaries
CREATE TABLE daily_summary
WITH (
  format = 'PARQUET',
  parquet_compression = 'SNAPPY',
  partitioned_by = ARRAY['year', 'month', 'day']
) AS
SELECT 
    user_id,
    event_type,
    COUNT(*) as event_count,
    SUM(amount) as total_amount,
    MAX(event_timestamp) as last_event
FROM raw_events
WHERE year = '2026'
GROUP BY user_id, event_type, year, month, day;

Athena Workgroup Configuration

{
  "Name": "analytics-team",
  "Configuration": {
    "ResultConfiguration": {
      "OutputLocation": "s3://athena-results/analytics/"
    },
    "EnforceWorkGroupConfiguration": true,
    "PublishCloudWatchMetrics": true,
    "BytesScannedCutoffPerQuery": 10737418240,
    "RequesterPaysEnabled": true
  }
}

Mathematical Formulas

Cost Optimization Formula

Throughput Calculation

Spot Savings

Performance Considerations

Optimization	Impact	Implementation
Adaptive Query Execution	2-5x faster	`spark.sql.adaptive.enabled=true`
Dynamic Allocation	30-50% cost reduction	`spark.dynamicAllocation.enabled=true`
Columnar Formats	2-10x faster reads	Parquet/ORC with Snappy compression
Partitioning	10-100x faster queries	Partition by date/region/business key
Predicate Pushdown	5-50x less data scanned	Filter at source, use partition pruning
Broadcast Joins	Avoid shuffle	`spark.sql.autoBroadcastJoinThreshold`
Coalesce/Repartition	Balanced parallelism	Match partition count to data volume

Security Considerations

Layer	Controls	Implementation
Network	VPC endpoints, private subnets	S3 Gateway endpoint, no public access
Identity	IAM roles, least privilege	Separate roles for EMR, Glue, Lambda
Data	Encryption at rest, TLS in transit	SSE-KMS for S3, TLS 1.2+ enforcement
Audit	CloudTrail, S3 access logging	Enable logging on all buckets and APIs
Secrets	Secrets Manager, rotation	Never hardcode credentials in scripts
Network Isolation	Security groups, NACLs	Restrict traffic between tiers

Interview Questions & Answers

Q1: How do you design a fault-tolerant batch processing pipeline on AWS?

Answer:

A fault-tolerant batch pipeline requires multiple layers of protection:

Checkpointing Strategy:

# EMR Spark checkpointing
spark.conf.set("spark.streaming.checkpointDirectory", 
               "s3://bucket/checkpoints/")

# Glue bookmarking
glueContext.create_dynamic_frame.from_catalog(
    database="mydb",
    table_name="mytable",
    transformation_ctx="datasource0"
)

Dead Letter Queue Pattern:

Route failed records to SQS DLQ
Alert on DLQ message count
Manual review and reprocessing

Retry Configuration:

{
  "Type": "Task",
  "Resource": "arn:aws:states:::elasticmapreduce:addStep.sync",
  "Retry": [
    {
      "ErrorEquals": ["States.TaskFailed"],
      "IntervalSeconds": 60,
      "MaxAttempts": 3,
      "BackoffRate": 2.0
    }
  ]
}

Q2: Compare EMR vs Glue for batch processing. When would you choose each?

Feature	EMR	Glue
Control	Full cluster control	Serverless, managed
Cost Model	EC2 instances	DPU-hours
Customization	Custom JARs, bootstrap	PySpark/Scala only
Use Case	Complex, long-running	Standard ETL

Choose EMR when: Running custom Spark/Flink applications, need specific instance types (GPU, memory-optimized), long-running clusters (hours/days), require fine-grained tuning.

Choose Glue when: Standard ETL transformations, want serverless auto-scaling, need built-in data catalog, short-running jobs (minutes).

Q3: How do you optimize Spark jobs on EMR for better performance?

1. Instance Selection:

aws emr create-cluster --instance-groups \
  InstanceGroupType=MASTER,InstanceCount=1,InstanceType=m6g.xlarge \
  InstanceGroupType=CORE,InstanceCount=4,InstanceType=m6g.2xlarge

2. Spark Configuration Tuning:

spark.sql.shuffle.partitions=200
spark.executor.memory=8g
spark.executor.cores=4
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=2
spark.dynamicAllocation.maxExecutors=50

3. Data Format Optimization:

Use Parquet with Snappy compression
Optimal partition size: 128MB-1GB per file
Cache frequently accessed DataFrames

Q4: Explain the Glue Job Bookmark mechanism and its benefits.

Glue Bookmarks track processed data to enable incremental processing:

datasource0 = glueContext.create_dynamic_frame.from_catalog(
    database="mydb",
    table_name="mytable",
    transformation_ctx="datasource0"
)

sink = glueContext.write_dynamic_frame.from_catalog(
    frame=apply_mapping,
    database="outputdb",
    table_name="outputtable",
    transformation_ctx="sink0"
)

Benefits: Eliminates duplicate processing, reduces runtime for incremental loads, no need for custom watermark tracking, automatic state management.

Q5: How do you handle schema evolution in batch pipelines?

Glue Schema Evolution:

sink = glueContext.write_dynamic_frame.from_catalog(
    frame=apply_mapping,
    database="outputdb",
    table_name="outputtable",
    enable_update_catalog=True,
    update_catalog_options={
        "updateBehavior": "UPDATE_IN_DATABASE",
        "schemaChangePolicy": "UPDATE"
    }
)

Spark Schema Evolution:

df = spark.read \
    .option("mergeSchema", "true") \
    .parquet("s3://bucket/data/")

Q6: Design a data quality framework for batch processing on AWS.

from pyspark.sql import DataFrame
from pyspark.sql.functions import col, count, when

def validate_data(df: DataFrame, rules: dict) -> dict:
    results = {}
    
    for col_name in rules.get('not_null', []):
        null_count = df.filter(col(col_name).isNull()).count()
        results[f'null_{col_name}'] = null_count == 0
    
    for col_name, (min_val, max_val) in rules.get('range', {}).items():
        out_of_range = df.filter(
            (col(col_name) < min_val) | (col(col_name) > max_val)
        ).count()
        results[f'range_{col_name}'] = out_of_range == 0
    
    return results

Q7: How do you implement idempotency in AWS batch jobs?

Deterministic Output Paths:

output_path = f"s3://bucket/output/year={year}/month={month}/day={day}/"
df.write.mode("overwrite").partitionBy("year", "month", "day").parquet(output_path)

Upsert Pattern with DynamoDB:

import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('processed_records')

def upsert_record(record):
    table.put_item(
        Item=record,
        ConditionExpression='attribute_not_exists(record_id) OR updated_at < :ts',
        ExpressionAttributeValues={':ts': record['updated_at']}
    )

Q8: Explain the cost optimization strategies for EMR clusters.

1. Instance Fleet with Spot (70% savings):

aws emr create-cluster --instance-fleets '[{
  "InstanceFleetType": "CORE",
  "TargetOnDemandCapacity": 2,
  "TargetSpotCapacity": 8
}]'

2. Auto-scaling Configuration:

{
  "ScaleOut": {
    "MetricName": "YARNMemoryAvailablePercentage",
    "TargetValue": 75.0,
    "ScaleOutCooldown": 300
  },
  "ScaleIn": {
    "MetricName": "YARNMemoryAvailablePercentage",
    "TargetValue": 30.0,
    "ScaleInCooldown": 600
  }
}

Common Pitfalls

Pitfall	Impact	Solution
Small files problem	Slow queries, high metadata overhead	Compact to 128MB-1GB files
Skewed partitions	One executor does most work	Repartition, use salting
No checkpointing	Full reprocess on failure	Enable bookmarks/checkpoints
Unoptimized joins	Shuffle, OOM errors	Broadcast small tables, co-join keys
Ignoring data skew	Imbalanced processing	Analyze distribution, salt keys
Over-provisioning	Wasted cost	Use dynamic allocation, spot instances
No monitoring	Silent failures	CloudWatch alarms, job metrics
Hardcoded configs	Brittle pipelines	Externalize configs, use SSM

AWS Batch Analytics for Data Engineers

AWS Batch Analytics for Data Engineers

Why This Matters

Batch Analytics Architecture

Real-World Project Structure

EMR for Batch Analytics

EMR Instance Selection for Batch Workloads

EMR Batch Job Production Code

EMR Cluster Creation with Cost Optimization

AWS Glue for Batch ETL

Glue ETL Job with Bookmarks

Athena for Ad-hoc Analytics

Athena Query Optimization

Athena Workgroup Configuration

Mathematical Formulas

Cost Optimization Formula

Throughput Calculation

Spot Savings

Performance Considerations

Security Considerations

Interview Questions & Answers

Q1: How do you design a fault-tolerant batch processing pipeline on AWS?

Q2: Compare EMR vs Glue for batch processing. When would you choose each?

Q3: How do you optimize Spark jobs on EMR for better performance?

Q4: Explain the Glue Job Bookmark mechanism and its benefits.

Q5: How do you handle schema evolution in batch pipelines?

Q6: Design a data quality framework for batch processing on AWS.

Q7: How do you implement idempotency in AWS batch jobs?

Q8: Explain the cost optimization strategies for EMR clusters.

Common Pitfalls

See Also

Premium Content