Chapter 35: Batch Processing

What is Batch Processing

Batch processing handles large volumes of data in discrete chunks, typically on a scheduled basis. Unlike stream processing which operates on data in real-time, batch systems process accumulated data all at once, optimizing for throughput over latency.

Common Use Cases

• ETL Pipelines: Extract data from sources, transform it, load into warehouses
• Analytics: Daily/hourly reports, aggregations, historical analysis
• ML Training: Process training datasets, feature engineering at scale
• Data Migration: One-time or periodic large-scale data movements
• Log Processing: Aggregate and analyze application logs

Batch vs Stream Processing

MapReduce

MapReduce, pioneered by Google in 2004, was the foundational programming model for distributed batch processing. While largely superseded by Spark, understanding MapReduce illuminates core distributed computing concepts.

Three-Phase Model

1. Map Phase: Process input splits in parallel, emit key-value pairs
2. Shuffle Phase: Partition, sort, and group intermediate data by key across network
3. Reduce Phase: Aggregate values for each key, write final output

Word Count Example

// Map function: emit (word, 1) for each word
map(String doc_id, String content):
    for each word w in content:
        emit(w, 1)

// Reduce function: sum counts for each word
reduce(String word, Iterator<Integer> counts):
    int sum = 0
    for each count in counts:
        sum += count
    emit(word, sum)

// Input: ["hello world", "hello hadoop"]
// Map output: [(hello, 1), (world, 1), (hello, 1), (hadoop, 1)]
// Shuffle groups: {hello: [1, 1], world: [1], hadoop: [1]}
// Reduce output: [(hello, 2), (world, 1), (hadoop, 1)]

Why MapReduce is Largely Replaced

• Disk I/O bottleneck: Writes intermediate results to disk between stages
• Rigid two-stage model: Complex workflows require chaining multiple jobs
• High latency: Job startup overhead, no interactive queries
• Limited expressiveness: Difficult to express iterative algorithms (ML)
• No in-memory processing: Cannot cache datasets for reuse

Spark addresses these limitations with in-memory computation, flexible DAGs, and richer APIs.

Apache Spark

Apache Spark is a unified analytics engine for large-scale data processing, offering up to 100x faster performance than MapReduce through in-memory computation and optimized execution plans.

Core Abstractions

• RDD (Resilient Distributed Dataset): Immutable, partitioned collection of records. Low-level API with transformations (map, filter) and actions (count, collect).
• DataFrame: Distributed table with named columns and schema. Higher-level API with SQL-like operations and Catalyst optimizer.
• Dataset: Typed DataFrame (Scala/Java), combines RDD type-safety with DataFrame optimizations.

Key Concepts

• Lazy Evaluation: Transformations build execution plan, actions trigger computation. Enables optimization.
• In-Memory Processing: Cache frequently accessed datasets in RAM across executors.
• Unified Engine: Single API for batch, SQL, streaming, ML, graph processing.

Spark Architecture

• Driver: Runs main() function, creates SparkContext, builds DAG, schedules tasks
• Cluster Manager: Allocates resources (YARN, Kubernetes, Mesos, Standalone)
• Executors: Run tasks, cache data, return results to driver

Example: DataFrame API

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum

spark = SparkSession.builder.appName("Sales").getOrCreate()

# Read Parquet files
sales = spark.read.parquet("s3://bucket/sales/")

# Lazy transformations (build execution plan)
result = sales \
    .filter(col("date") >= "2024-01-01") \
    .groupBy("product_id") \
    .agg(sum("revenue").alias("total_revenue")) \
    .orderBy(col("total_revenue").desc())

# Action triggers execution
result.show(10)  # Display top 10
result.write.parquet("s3://bucket/results/")  # Persist

Spark Execution Model

Understanding how Spark translates high-level code into distributed execution is critical for optimization.

DAG of Stages

Spark compiles transformations into a Directed Acyclic Graph (DAG) of stages. Stage boundaries occur at shuffle operations (wide transformations).

Narrow vs Wide Transformations

Narrow transformations can be pipelined in a single stage. Wide transformations require shuffle, creating stage boundaries.

Partitioning

Data is divided into partitions, the unit of parallelism. Each partition is processed by one task on one executor core.

• Default partitioning: Based on input file blocks (typically 128MB HDFS blocks)
• Hash partitioning: partition = hash(key) % numPartitions
• Range partitioning: Partitions cover ranges of sorted keys
• Custom partitioning: User-defined partitioner for domain-specific logic

Execution Flow

1. Logical Plan: High-level DataFrame/SQL operations
2. Optimized Plan: Catalyst optimizer applies rules (predicate pushdown, constant folding)
3. Physical Plan: Concrete execution strategy (broadcast join vs shuffle join)
4. DAG of Stages: Split at shuffle boundaries
5. Task Scheduling: Driver assigns tasks to executors (data locality preferred)
6. Execution: Tasks process partitions, shuffle if needed, return results

// Example: view execution plan
df.explain()  // Shows logical and physical plans
df.explain(mode="extended")  // Detailed optimization steps

Shuffle Optimization

Shuffle is the most expensive operation in Spark, involving disk writes, network transfer, and sorting. Optimizing shuffle is key to performance.

Why Shuffle is Expensive

• Disk I/O: Map tasks write shuffle files to disk, reduce tasks read them
• Network Transfer: Data moves across executors (potentially cross-rack/datacenter)
• Serialization: Data must be serialized for network transfer, deserialized on receive
• Sorting: Shuffle often involves sorting data by key
• Straggler Tasks: Data skew causes some tasks to process disproportionate data

Optimization Strategies

1. Partition Pruning

Reduce shuffle volume by filtering before wide transformations.

// Bad: shuffle entire dataset, then filter
df.groupBy("user_id").agg(sum("amount")).filter(col("date") == "2024-01-15")

// Good: filter first, shuffle less data
df.filter(col("date") == "2024-01-15").groupBy("user_id").agg(sum("amount"))

2. Broadcast Joins

When joining a large table with a small table (< 10MB default), broadcast the small table to all executors. Avoids shuffling the large table.

from pyspark.sql.functions import broadcast

# Broadcast small dimension table to avoid shuffle
large_df.join(broadcast(small_df), "key")

3. Bucketing

Pre-partition and sort tables on join keys during write. Future joins on same key avoid shuffle.

// Write bucketed table
df.write \
  .bucketBy(100, "user_id") \
  .sortBy("user_id") \
  .saveAsTable("users_bucketed")

// Join bucketed tables - no shuffle needed
users_bucketed.join(orders_bucketed, "user_id")

4. Salting for Skew

When keys are heavily skewed (e.g., 80% of data has same key), add random salt to distribute load.

from pyspark.sql.functions import rand, concat, lit

# Add salt to skewed key
df_salted = df.withColumn("salt", (rand() * 10).cast("int")) \
              .withColumn("salted_key", concat(col("skewed_key"), lit("_"), col("salt")))

# Aggregate with salted key
result = df_salted.groupBy("salted_key").agg(sum("value"))

# Remove salt and aggregate again
final = result.withColumn("original_key",
    split(col("salted_key"), "_")[0]).groupBy("original_key").agg(sum("value"))

5. Coalesce vs Repartition

• coalesce(n): Reduce partitions without shuffle (combines adjacent partitions). Use when reducing partition count.
• repartition(n): Full shuffle to exactly n partitions. Use when increasing partitions or when even distribution is critical.

6. Tune Shuffle Partitions

Default shuffle partitions (200) may be too small for large data or too large for small data.

spark.conf.set("spark.sql.shuffle.partitions", "1000")  # Increase for large data
spark.conf.set("spark.sql.adaptive.enabled", "true")  # Auto-tune partitions (Spark 3.0+)

Data Formats

Choosing the right storage format significantly impacts batch processing performance. Columnar formats (Parquet, ORC) are preferred for analytics.

Format Comparison

Parquet (Recommended for Analytics)

• Columnar storage: Read only needed columns, skip irrelevant data
• Predicate pushdown: Filter at read time using statistics (min/max, null counts)
• Compression: Columnar data compresses better (same type, patterns)
• Encoding: Dictionary encoding, run-length encoding for repeated values
• Nested types: Supports complex schemas (structs, arrays, maps)

Delta Lake (ACID on Data Lake)

Delta Lake adds transactional guarantees to Parquet files, enabling reliable upserts and time travel.

• ACID transactions: Atomic commits, isolation from concurrent readers/writers
• Time travel: Query historical versions via transaction log
• Schema enforcement: Reject writes that violate schema
• Upserts/Deletes: Merge operations for change data capture (CDC)
• Compaction: Optimize small files into larger files

from delta.tables import DeltaTable

# Write Delta table
df.write.format("delta").save("/data/events")

# Upsert (merge)
deltaTable = DeltaTable.forPath(spark, "/data/events")
deltaTable.alias("target").merge(
    updates.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(set = {"status": "updated"}) \
 .whenNotMatchedInsert(values = {...}) \
 .execute()

# Time travel
spark.read.format("delta").option("versionAsOf", 5).load("/data/events")

Orchestration

Batch jobs rarely run in isolation. Orchestration systems manage dependencies, scheduling, retries, and monitoring across complex pipelines.

Apache Airflow

Airflow defines workflows as Python code (DAGs), with rich operators for common tasks and strong community support.

from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-eng',
    'retries': 3,
    'retry_delay': timedelta(minutes=5)
}

with DAG('sales_etl',
         default_args=default_args,
         schedule_interval='@daily',
         start_date=datetime(2024, 1, 1),
         catchup=False) as dag:

    extract = SparkSubmitOperator(
        task_id='extract_sales',
        application='/jobs/extract.py',
        conf={'spark.executor.memory': '4g'}
    )

    transform = SparkSubmitOperator(
        task_id='transform_sales',
        application='/jobs/transform.py'
    )

    load = SparkSubmitOperator(
        task_id='load_warehouse',
        application='/jobs/load.py'
    )

    notify = PythonOperator(
        task_id='send_notification',
        python_callable=send_slack_message
    )

    extract >> transform >> load >> notify  # Define dependencies

Key Orchestration Features

• DAG-based dependencies: Define task execution order and parallelism
• Scheduling: Cron expressions, catchup for backfills
• Retries: Automatic retry with exponential backoff
• Sensors: Wait for external events (file arrival, API availability)
• SLAs: Alert if tasks exceed time limits
• Monitoring: Web UI for status, logs, metrics
• Dynamic DAGs: Generate tasks programmatically

Alternatives

• Dagster: Asset-oriented (tracks datasets, not just tasks), strong testing support, type-safe
• Prefect: Python-native, hybrid execution (cloud/on-prem), simpler than Airflow
• Temporal: Workflow-as-code, durable execution, handles long-running workflows
• AWS Step Functions: Serverless orchestration for AWS services

Fault Tolerance

Batch jobs can run for hours on thousands of nodes. Fault tolerance mechanisms ensure job completion despite node failures.

Spark Fault Tolerance Mechanisms

1. Lineage-Based Recovery

Spark tracks the lineage (DAG) of RDDs. If a partition is lost, Spark recomputes it from the source data using the lineage graph.

• No replication overhead: Unlike HDFS replication, lineage doesn't duplicate data
• Fine-grained recovery: Only lost partitions are recomputed, not entire dataset
• Works for narrow transformations: Lost partition can be recomputed from parent partition

2. Checkpointing

For long lineages (e.g., iterative algorithms), recomputation can be expensive. Checkpointing saves RDD to reliable storage, truncating lineage.

rdd.checkpoint()  // Save RDD to HDFS/S3
spark.sparkContext.setCheckpointDir("s3://bucket/checkpoints/")

Use when: (1) lineage grows very long (many transformations), (2) shuffles are expensive to recompute, (3) iterative algorithms (ML training).

3. Speculative Execution

If a task is running significantly slower than others (stragglers), Spark launches duplicate task on another node. First to finish wins.

spark.conf.set("spark.speculation", "true")
spark.conf.set("spark.speculation.multiplier", "1.5")  // 1.5x median time triggers speculation

Helps with: slow nodes, data skew, noisy neighbors in cloud.

4. Task Retries

Failed tasks are automatically retried (default 4 attempts). If a task fails repeatedly, the entire stage fails, and driver can decide to retry or abort.

5. Dynamic Allocation

Spark can request more executors when tasks are pending, and release idle executors. Helps with node failures and resource efficiency.

spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=5
spark.dynamicAllocation.maxExecutors=100

Best Practices

• Idempotent operations: Rerunning should produce same output (use overwrite mode, not append)
• Persist intermediate results: Cache or checkpoint expensive computations
• Monitor stragglers: Use Spark UI to identify slow tasks, optimize data skew
• Set appropriate retries: Balance fault tolerance vs. wasted compute

Resource Management

Batch processing systems must share cluster resources efficiently. Cluster managers allocate CPU, memory, and schedule tasks across nodes.

Cluster Manager Options

1. YARN (Hadoop)

YARN (Yet Another Resource Negotiator) is the resource manager for Hadoop ecosystems.

• ResourceManager: Global scheduler, allocates containers to applications
• NodeManager: Per-node agent, launches and monitors containers
• ApplicationMaster: Per-application, negotiates resources, manages tasks
• Queues: Multi-tenancy with capacity/fair scheduler, preemption

2. Kubernetes

Kubernetes (K8s) is increasingly popular for Spark, offering containerization and cloud-native benefits.

• Dynamic scaling: Auto-scale executors based on workload
• Isolation: Containers provide better resource isolation than YARN
• Multi-tenancy: Namespaces, resource quotas, network policies
• Cloud integration: Native support in GKE, EKS, AKS

spark-submit \
  --master k8s://https://<k8s-apiserver> \
  --deploy-mode cluster \
  --conf spark.executor.instances=10 \
  --conf spark.kubernetes.container.image=spark:3.5.0 \
  --conf spark.kubernetes.namespace=data-eng \
  local:///app/job.py

3. Standalone

Spark's built-in cluster manager. Simple, but lacks advanced features (multi-tenancy, resource quotas).

4. Mesos

General-purpose cluster manager (pre-Kubernetes era). Less common now.

Dynamic Allocation

Spark can scale executors up/down based on workload, improving resource utilization.

• Scale up: Add executors when tasks are pending
• Scale down: Remove executors idle for > spark.dynamicAllocation.executorIdleTimeout
• External shuffle service: Required for dynamic allocation (preserves shuffle files after executor dies)

Resource Sizing Guidelines

spark-submit \
  --executor-memory 6G \
  --executor-cores 5 \
  --num-executors 20 \
  --driver-memory 2G \
  job.py

Alternatives to Spark

While Spark dominates batch processing, alternatives excel in specific scenarios.

Comparison Table

Apache Flink

Flink treats batch as a special case of streaming (bounded stream). Excels at stateful stream processing with exactly-once semantics.

• True streaming: Event-time processing, watermarks, late data handling
• Low latency: Millisecond processing (vs. Spark's micro-batches)
• Savepoints: Consistent snapshots for upgrades, migrations

Apache Beam

Beam provides a unified API for batch and stream processing, with pluggable runners (Spark, Flink, Dataflow).

• Portability: Write once, run on any runner
• Unified model: Same API for batch and stream
• Cloud integration: Native Dataflow runner on GCP

Presto/Trino

Distributed SQL query engine for interactive analytics. Not a full batch processing system, but excellent for ad-hoc queries on data lakes.

• Query federation: Join across multiple sources (S3, HDFS, Postgres, Kafka)
• Low latency: Interactive queries (seconds), not hours like Hive
• No data movement: Query in-place, no ETL into warehouse

When to Use Batch Processing

Choosing between batch, stream, and hybrid approaches depends on latency requirements, data volume, and processing complexity.

Decision Criteria

Choose Batch When:

• Latency is flexible: Hourly, daily, or weekly results are acceptable
• Large historical data: Processing terabytes to petabytes of historical data
• Complex transformations: Multi-stage ETL, ML model training, graph algorithms
• Cost-sensitive: Batch can use spot/preemptible instances, shutdown when idle
• Scheduled workloads: Reports, aggregations, backups run on fixed schedule
• Complete data required: Global sorts, aggregations across entire dataset

Choose Stream When:

• Low latency critical: Real-time alerts, fraud detection, monitoring
• Continuous processing: Data arrives continuously, insights needed immediately
• Stateful operations: Sessionization, windowed aggregations, pattern detection
• Event-driven: Reacting to events as they occur (clickstream, IoT sensors)

Hybrid (Lambda/Kappa):

• Lambda: Batch layer for accuracy + stream layer for low latency. Combine results. Complex but comprehensive.
• Kappa: Stream-only, reprocess from Kafka log for corrections. Simpler but requires replayable source.

Scaling Considerations

Example Decision Tree