Batch Processing at Scale: Spark & Flink Patterns - Data Engineering Hub

← Back to Hub Batch Processing at Scale: Spark & Flink Patterns Production patterns for petabyte-scale ETL using Apache Spark, Flink, and Apache Beam. Covers partitioning strategies, join optimization, shuffle reduction, and cost-efficient cluster management. Table of Contents Execution Model Comparison Partitioning Strategies Join Optimization Shuffle Reduction Cost-Efficient Cluster Management Flink vs Spark: When to Use Each Production Patterns Checklist Execution Model Comparison Understanding the fundamental execution differences between batch engines determines which patterns apply: Apache Spark (Micro-Batch) Model: RDD/Dataset lineage graphs with lazy evaluation Scheduling: DAGScheduler breaks jobs into stages at shuffle boundaries Shuffle: Sort-based shuffle with spill-to-disk; Tungsten encoder for off-heap State: Stateless transformations; stateful via mapGroupsWithState or flatMapGroupsWithState Best for: Iterative ML, complex SQL, broad ecosystem, existing Hive metastore Apache Flink (Native Streaming) Model: Continuous dataflow with event-time processing and watermarks Scheduling: Pipelined execution; no stage boundaries for pure streaming State: First-class keyed state (RocksDB backend) with exactly-once checkpoints Shuffle: Network stack with credit-based flow control; pipelined shuffle Best for: Low-latency streaming, complex event processing, exactly-once requirements Apache Beam (Unified Model) Model: Portable runner API; same pipeline runs on Spark, Flink, Dataflow, Samza Abstraction: PCollection + PTransform with windowing, triggers, accumulation modes Trade-off: Runner-specific optimization gaps; prefer native for complex tuning Partitioning Strategies Partitioning determines parallelism, data locality, and shuffle volume. Three complementary approaches: 1. Input Partitioning (Source-Side) # Spark: Optimal partition count = 2-3x CPU cores spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB") # File-based: maxPartitionBytes + openCostInBytes spark.read.option("maxPartitionBytes", "128MB").parquet("s3://bucket/data/") # JDBC: partitionColumn + lowerBound/upperBound + numPartitions spark.read.jdbc(url, table, column="id", lowerBound=1, upperBound=1e9, numPartitions=200) 2. Data Partitioning (Shuffle-Side) # Repartition by business key BEFORE wide transformations df = df.repartition("customer_id") # Co-locates customer data # Spark AQE (Adaptive Query Execution) handles skew automatically spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true") spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5") spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB") # Custom partitioner for composite keys df.rdd.partitionBy(HashPartitioner(200), lambda x: (x[0], x[1] % 10)) 3. Output Partitioning (Sink-Side) # Hive-style partitioning for predicate pushdown df.write.partitionBy("year", "month", "day").parquet("s3://warehouse/events/") # Bucketing for join optimization (Spark) df.write.bucketBy(200, "user_id").sortBy("event_time").saveAsTable("events_bucketed") # Flink: Custom partitioner for sinks sink.addSink(new StreamingFileSink...) Join Optimization Joins are the primary shuffle driver. Choose strategy by data size and skew: Strategy Condition Config Broadcast Hash Join Small table < 100MB (configurable) spark.sql.autoBroadcastJoinThreshold Sort-Merge Join Large tables, equi-join Default for > threshold Shuffle Hash Join No sort needed, build side fits memory spark.sql.join.preferSortMergeJoin=false Cartesian Product Cross join (avoid!) spark.sql.crossJoin.enabled=true Skew Handling Patterns # 1. Salt the skewed key (split hot keys) from pyspark.sql.functions import col, rand, concat, lit df_salted = df.withColumn("salted_key", concat(col("skewed_col"), lit("_"), (rand() * 100).cast("int"))) df_salted.join(other_salted, "salted_key").drop("salted_key") # 2. AQE skew join (auto-split skewed partitions) # Enable: spark.sql.adaptive.skewJoin.enabled=true # 3. Flink: Keyed state with custom aggregation for hot keys # Use State TTL to expire stale hot-key state stateTTLConfig = StateTtlConfig.newBuilder(Time.minutes(30)) .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite) .build() Shuffle Reduction Techniques Every shuffle writes to disk and reads over network. Eliminate or minimize: 1. Filter Early, Project Early # BAD: Shuffle full rows then filter df.join(other, "id").filter(col("status") == "active").select("id", "name") # GOOD: Push predicates before join df.filter(col("status") == "active").select("id", "name").join(other.select("id"), "id") 2. Map-Side Pre-Aggregation (Combiner Pattern) # Spark: reduceByKey / aggregateByKey before shuffle rdd.map(lambda x: (x.key, x.value)) .reduceByKey(lambda a, b: a + b) # Local pre-aggregation .map(lambda x: process(x)) # Single shuffle # Flink: AggregateFunction with local pre-aggregation class CountAgg implements AggregateFunction { public Long createAccumulator() { return 0L; } public Long add(Event e, Long acc) { return acc + 1; } public Long getResult(Long acc) { return acc; } public Long merge(Long a, Long b) { return a + b; } // Called after local pre-agg } 3. Broadcast Variables for Lookup Joins # Spark: Broadcast small dimension tables broadcast_dim = spark.sparkContext.broadcast(dim_df.collectAsMap()) result = fact_df.rdd.map(lambda row: enrich(row, broadcast_dim.value)) # Flink: Broadcast State Pattern (for dynamic config) BroadcastStateDescriptor descriptor = new BroadcastStateDescriptor<>("config", StringSerializer.INSTANCE, ConfigSerializer.INSTANCE); configStream.broadcast(descriptor).connect(eventStream).process(new EnrichmentFunction()); 4. Column Pruning & Predicate Pushdown # Parquet/ORC: Only read needed columns + push predicates to scan spark.read.parquet("s3://data/").select("id", "timestamp", "value") .filter(col("timestamp") > "2024-01-01") # Pushed to Parquet footer stats Cost-Efficient Cluster Management Spot/Preemptible Instances with Graceful Decomposition # Spark: Decommission executors gracefully on spot interruption spark.conf.set("spark.executor.decommission.enabled", "true") spark.conf.set("spark.executor.decommission.timeout", "300s") # Kubernetes: PodDisruptionBudget for controlled eviction apiVersion: policy/v1 kind: PodDisruptionBudget metadata: name: spark-executor-pdb spec: minAvailable: 50% selector: matchLabels: spark-role: executor Dynamic Allocation (Spark) spark.conf.set("spark.dynamicAllocation.enabled", "true") spark.conf.set("spark.dynamicAllocation.minExecutors", "10") spark.conf.set("spark.dynamicAllocation.maxExecutors", "200") spark.conf.set("spark.dynamicAllocation.initialExecutors", "50") spark.conf.set("spark.dynamicAllocation.executorIdleTimeout", "120s") spark.conf.set("spark.dynamicAllocation.cachedExecutorIdleTimeout", "300s") Flink: Reactive Scaling with Savepoints # Savepoint-triggered rescaling (manual or via Flink Kubernetes Operator) flink savepoint :jobid s3://bucket/savepoints/ flink rescale -r 100 :jobid # Rescale to 100 parallelism # Native Kubernetes: FlinkDeployment with autoscaling apiVersion: flink.apache.org/v1beta1 kind: FlinkDeployment spec: job: parallelism: 50 upgradeMode: savepoint autoscaling: enabled: true minParallelism: 10 maxParallelism: 200 Flink vs Spark: Decision Matrix Requirement Choose Spark Choose Flink Latency SLA Minutes-hours OK Seconds-milliseconds required Exactly-once Idempotent sinks + transactions Native end-to-end Stateful ops Limited (mapGroupsWithState) First-class keyed state Event time Watermarks in Structured Streaming Native, flexible watermarks Iterative ML Excellent (caching, MLlib) Possible but less mature SQL/BI Spark SQL + Thrift server Flink SQL (improving) Team skills Wider talent pool Specialized Production Patterns Checklist Monitoring & Observability Spark: Spark UI history server + Prometheus JMX exporter + custom metrics via SparkListener Flink: Flink metrics (Prometheus reporter) + REST API for job/vertex/task metrics Key metrics: Shuffle read/write bytes, GC time %, task skew, input/output records, checkpoint duration (Flink) Data Quality Gates # Spark: Assertions as DataFrame actions (fail fast) df.filter(col("id").isNull()).count() match { case 0 => // OK case n => throw new RuntimeException(s"Null IDs: $n") } # Flink: Side outputs for invalid records OutputTag invalidTag = new OutputTag<>("invalid"){}; SingleOutputStreamOperator valid = events.process(new ValidationFunction()) .getSideOutput(invalidTag); // Capture for dead-letter queue Idempotent Sinks for Reprocessing # Upsert pattern with deduplication key df.write .format("delta") // or Iceberg, Hudi .option("mergeSchema", "true") .mode("overwrite") .option("replaceWhere", "date >= '2024-01-01' AND date <= '2024-01-31'") .save("s3://warehouse/table/") # Flink: Two-phase commit sink for exactly-once public class ExactlyOnceSink implements TwoPhaseCommitSink { // beginTransaction, invoke, commit, recoverAndCommit, abort } Testing Strategies Unit: Spark: spark-testing-base / Flink: TestHarness for operators Integration: Mini clusters (Spark LocalSparkContext , Flink MiniClusterWithClientResource ) Contract: Schema registry (Confluent/Avro) + compatibility checks in CI Data: Great Expectations / Soda Core for pipeline-level assertions Related Guides Real-Time Streaming with Kafka & Flink Data Warehouse Architecture Data Quality Contracts Modern Orchestration