Traffic Spike Scenario: 50K/s → 500K/s in 90 Seconds

Events jump from 50K/s to 500K/s in 90 seconds. What breaks first — and how do you absorb the shock without dropping events?

The Setup

You have a real-time streaming pipeline processing payment events:

Event Producers (Mobile Apps, Web, IoT)
        │
        ▼
   Apache Kafka (Message Bus)
        │
        ▼
   Spark Structured Streaming (Consumer)
        │
        ▼
   Data Lake (S3/ADLS) + Real-Time Dashboard

Under normal load, your pipeline processes 50,000 events per second comfortably. Then a flash sale starts, and within 90 seconds, the event rate surges to 500,000 events per second — a 10x spike.

What Breaks First?

Understanding the failure cascade is critical. Components do NOT fail simultaneously — they fail in a predictable sequence:

graph TD
    A["10x Traffic Spike Hits"] --> B["Spark micro-batch processing time exceeds trigger interval"]
    B --> C["Consumer lag grows on Kafka"]
    C --> D["Executor memory fills up (GC pressure)"]
    D --> E["OOM kills executors / tasks fail"]
    E --> F["Spark retries tasks → more pressure"]
    F --> G["Downstream dashboards show stale data"]
    G --> H["Kafka retention expires → EVENTS LOST"]

Failure Sequence (in order):

Key Insight: The Spark consumer is always the first bottleneck, not Kafka. Kafka is designed to handle massive throughput — it's a dumb pipe. The intelligence (and fragility) lives in the consumer.

The Defense: A 5-Layer Absorption Strategy

Layer 1: Kafka as the Shock Absorber

Kafka's core value during spikes is its durable, high-retention buffer. It decouples the producer write speed from the consumer read speed.

Critical configurations:

# Kafka Broker Configuration
log.retention.hours=168              # 7 days retention (not the default 24h!)
log.retention.bytes=-1               # No size-based deletion
num.partitions=120                   # High partition count for parallelism
replica.factor=3                     # Data durability

# Kafka Producer Configuration (event source side)
acks=all                             # Wait for all replicas before ACK
buffer.memory=67108864               # 64 MB producer buffer
batch.size=65536                     # 64 KB batch for throughput
linger.ms=10                         # Allow 10ms batching window
compression.type=lz4                 # Compress to reduce network IO

Why 7 days retention? If your consumer goes down for 6 hours (or even a full day), you want the data to still be sitting in Kafka when it comes back. The default 24h retention is dangerously short for production systems.

Why 120 partitions? During the spike, you need Spark to scale to many executors. Each Spark task reads from one Kafka partition. If you have only 10 partitions, you can never run more than 10 parallel tasks — no matter how many executors you add.

Normal (50K/s):        10 partitions × 5K/s per partition = 50K/s ✅
Spike  (500K/s):       10 partitions × 50K/s per partition = 50K/s per task ❌ (overloaded)
Spike  (500K/s):      120 partitions × ~4.2K/s per partition = 500K/s ✅ (distributed)

Layer 2: Spark Backpressure & Rate Limiting

Spark Structured Streaming has a built-in backpressure mechanism called maxOffsetsPerTrigger. This caps the number of records Spark reads from Kafka per micro-batch, preventing the consumer from being overwhelmed.

# PySpark Structured Streaming with backpressure
stream_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092") \
    .option("subscribe", "payment_events") \
    .option("maxOffsetsPerTrigger", 1000000)   \  # Max 1M records per micro-batch
    .option("startingOffsets", "latest") \
    .option("failOnDataLoss", "false") \
    .load()

How it works during a spike:

Normal (50K/s, trigger every 10s):
    Records available per trigger: 500,000
    maxOffsetsPerTrigger: 1,000,000
    Records processed: 500,000 (all available) ✅

Spike (500K/s, trigger every 10s):
    Records available per trigger: 5,000,000
    maxOffsetsPerTrigger: 1,000,000
    Records processed: 1,000,000 (capped!) ✅
    Remaining 4,000,000 stay in Kafka for next batch.

Trade-off: Rate limiting increases consumer lag (data is delayed), but it prevents OOM crashes and data loss. Lag is recoverable; OOM crashes are not.

Layer 3: Dynamic Resource Allocation (Auto-Scaling)

Spark's Dynamic Resource Allocation lets the cluster automatically add executors when the workload increases and release them when it decreases.

# SparkSession configuration for auto-scaling
spark = SparkSession.builder \
    .appName("PaymentEventProcessor") \
    .config("spark.dynamicAllocation.enabled", "true") \
    .config("spark.dynamicAllocation.minExecutors", 5) \
    .config("spark.dynamicAllocation.maxExecutors", 50) \
    .config("spark.dynamicAllocation.executorIdleTimeout", "120s") \
    .config("spark.dynamicAllocation.schedulerBacklogTimeout", "5s") \
    .config("spark.dynamicAllocation.sustainedSchedulerBacklogTimeout", "5s") \
    .config("spark.shuffle.service.enabled", "true") \
    .getOrCreate()

Scaling timeline during a spike:

Time 0s:     Normal load. 5 executors active.
Time 0-90s:  Spike builds. Scheduler detects pending tasks backlog.
Time 95s:    schedulerBacklogTimeout (5s) exceeded → requests 10 more executors.
Time 100s:   10 more executors requested → requests another 15.
Time 120s:   30 executors active. Processing catches up.
Time 300s:   Spike subsides. Executors idle for 120s → released.
Time 420s:   Back to 5 executors.

Layer 4: Micro-Batch Tuning

The trigger interval determines how frequently Spark polls Kafka for new data:

# Trigger interval tuning
query = stream_df.writeStream \
    .trigger(processingTime="30 seconds") \   # Process every 30 seconds
    .outputMode("append") \
    .format("delta") \
    .option("checkpointLocation", "/checkpoints/payments") \
    .start("/data/payments_processed")

Recommendation: During spikes, a 30-second trigger interval with maxOffsetsPerTrigger gives the best balance between throughput and stability.

Layer 5: Monitoring & Alerting

You need to detect the spike BEFORE it crashes your pipeline:

# Monitor consumer lag via Kafka metrics
from pyspark.sql import functions as F

# In each micro-batch, track processing metrics
def log_batch_metrics(batch_df, batch_id):
    record_count = batch_df.count()
    print(f"[Batch {batch_id}] Records processed: {record_count}")

    # Alert if batch size exceeds 2x normal
    if record_count > 1_000_000:
        send_alert(f"⚠️ SPIKE DETECTED: Batch {batch_id} has {record_count} records (>1M threshold)")

query = stream_df.writeStream \
    .foreachBatch(log_batch_metrics) \
    .trigger(processingTime="30 seconds") \
    .start()

Key Metrics to Monitor:

The Complete Production Configuration

Putting it all together — here's the production-ready Spark Structured Streaming config for spike absorption:

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder \
    .appName("SpikeResilientPaymentProcessor") \
    .config("spark.dynamicAllocation.enabled", "true") \
    .config("spark.dynamicAllocation.minExecutors", 5) \
    .config("spark.dynamicAllocation.maxExecutors", 50) \
    .config("spark.dynamicAllocation.schedulerBacklogTimeout", "5s") \
    .config("spark.dynamicAllocation.executorIdleTimeout", "120s") \
    .config("spark.shuffle.service.enabled", "true") \
    .config("spark.executor.memory", "8g") \
    .config("spark.executor.cores", 4) \
    .config("spark.sql.shuffle.partitions", 200) \
    .config("spark.streaming.backpressure.enabled", "true") \
    .getOrCreate()

# Read from Kafka with rate limiting
raw_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092") \
    .option("subscribe", "payment_events") \
    .option("maxOffsetsPerTrigger", 2000000) \
    .option("startingOffsets", "latest") \
    .option("failOnDataLoss", "false") \
    .option("kafka.max.partition.fetch.bytes", "10485760") \
    .load()

# Parse and process
parsed = raw_stream.select(
    F.from_json(F.col("value").cast("string"), schema).alias("event"),
    F.col("timestamp").alias("kafka_timestamp"),
    F.col("partition"),
    F.col("offset")
).select("event.*", "kafka_timestamp", "partition", "offset")

# Write with checkpointing
query = parsed.writeStream \
    .trigger(processingTime="30 seconds") \
    .outputMode("append") \
    .format("delta") \
    .option("checkpointLocation", "s3://pipeline/checkpoints/payments") \
    .option("mergeSchema", "true") \
    .start("s3://datalake/payments/")

query.awaitTermination()

Summary: The Spike Absorption Playbook

The Golden Rule: Kafka absorbs the shock, rate-limiting prevents the crash, auto-scaling catches up, and monitoring tells you it happened. No events are dropped.

Follow-Up Questions & Answers

Q1: What if the spike is sustained for 48 hours instead of 90 seconds?

A: A 90-second spike is a burst — Kafka absorbs it and Spark catches up. A 48-hour sustained 10x load is a capacity problem, not a spike. The response changes fundamentally:

• Short-term (first 2 hours): Your auto-scaling hits maxExecutors and stays there. Consumer lag stabilizes but doesn't decrease. You're in a steady-state deficit.
• Action required: You must increase the cluster ceiling — either raise maxExecutors from 50 to 150, or vertically scale by switching from G.1X to G.2X workers.
• Long-term: If 10x is the new normal, you need to permanently resize the pipeline — more Kafka partitions, more baseline executors, and a higher maxOffsetsPerTrigger.

Spike (temporary):    Scale UP → absorb → scale DOWN
Sustained load shift: Scale UP → stay up → re-baseline capacity permanently

Q2: What if Kafka itself becomes the bottleneck (brokers can't handle 500K/s writes)?

A: This happens when your Kafka cluster is undersized. The producer starts getting TimeoutException or RecordTooLargeException. Events are dropped at the source before they even reach your pipeline.

Fixes (in order of urgency):

Key metric to watch: Kafka broker's UnderReplicatedPartitions. If this goes above 0, your brokers are struggling.

Q3: We use Spark on Kubernetes. Dynamic Allocation doesn't work the same way — what do we do?

A: On Kubernetes, Spark's Dynamic Allocation needs the External Shuffle Service or shuffle tracking (spark.dynamicAllocation.shuffleTracking.enabled=true in Spark 3.0+). Without it, Spark can't release executors because shuffle data would be lost.

# Spark on K8s with dynamic allocation
spark.conf.set("spark.dynamicAllocation.enabled", "true")
spark.conf.set("spark.dynamicAllocation.shuffleTracking.enabled", "true")  # K8s-specific!
spark.conf.set("spark.kubernetes.allocation.batch.size", 10)  # Request 10 pods at a time
spark.conf.set("spark.kubernetes.allocation.batch.delay", "5s")

Alternatively, use Horizontal Pod Autoscaler (HPA) on the Kubernetes level, scaling the Spark executor pods based on CPU/memory metrics. This is infrastructure-level scaling rather than Spark-level.

Q4: What if the spike causes failOnDataLoss=true to crash the job because some Kafka offsets expired?

A: This is exactly why we set failOnDataLoss=false. But the interviewer is asking: what do you lose when you set it to false?

• With failOnDataLoss=true (default): Spark crashes if the checkpoint references an offset that no longer exists in Kafka (retention expired). This is a safety mechanism — it tells you data was lost.
• With failOnDataLoss=false: Spark silently skips the missing offsets and continues from the earliest available offset. You lose data without any exception being thrown.

The production answer: Set failOnDataLoss=false BUT add a custom monitor that tracks the gap:

def detect_data_loss(batch_df, batch_id):
    # Compare the earliest offset we read vs what the checkpoint expected
    min_offset = batch_df.agg(F.min("offset")).collect()[0][0]
    expected_offset = get_checkpoint_offset()  # Read from checkpoint metadata

    if min_offset > expected_offset:
        gap = min_offset - expected_offset
        send_alert(f"🚨 DATA LOSS DETECTED: {gap} offsets skipped in partition!")

Q5: How do you handle a spike that comes with schema changes (e.g., a new field added during the surge)?

A: This is a compound problem — spike + schema evolution simultaneously.

• If using Delta Lake: Enable mergeSchema=true. New columns are added automatically without breaking the pipeline.
• If using Parquet/plain S3: The pipeline crashes on schema mismatch. You need to handle it explicitly:

# Defensive schema handling during spikes
try:
    parsed = raw_stream.select(
        F.from_json(F.col("value").cast("string"), expected_schema).alias("event")
    )
except Exception:
    # Fallback: use permissive mode
    parsed = raw_stream.select(
        F.from_json(F.col("value").cast("string"), expected_schema, 
                    {"mode": "PERMISSIVE"}).alias("event")
    )

Best practice: Always use PERMISSIVE mode for JSON parsing in streaming pipelines. It puts unparseable fields in a _corrupt_record column rather than crashing.

Sub-Scenarios

Sub-Scenario A: The Spike Hits at 3 AM With No Engineers On-Call

Situation: The spike happens at 3 AM. The consumer lag grows to 5 billion messages. The PagerDuty fires but nobody responds for 2 hours.

What should have been in place:

1. Auto-scaling should handle it automatically — no human needed for the first 30 minutes.
2. Auto-restart on OOM: Configure spark.yarn.maxAppAttempts=3 so the job restarts itself after a crash.
3. Escalation policy: If lag exceeds 10M for > 15 minutes, auto-escalate from Slack → PagerDuty → Phone call.
4. Self-healing runbook: An automated script that detects sustained lag and increases maxExecutors via the YARN/K8s API.

Sub-Scenario B: Spike + Data Skew Simultaneously

Situation: The 10x spike is caused by a single hot key (one mega-merchant generating all the extra traffic). Now you have both a volume problem AND a skew problem.

The fix stack:

Layer 1: Kafka absorbs the volume
Layer 2: maxOffsetsPerTrigger caps per-batch load
Layer 3: Auto-scaling adds executors
Layer 4: Salting breaks the hot key across executors  ← NEW for skew!
Layer 5: AQE splits skewed partitions at runtime     ← NEW for skew!

Without the salting/AQE layers, the extra executors added by auto-scaling are USELESS — all traffic still routes to the same one executor that owns the hot key's hash partition.

Sub-Scenario C: Spike During a Cluster Maintenance Window

Situation: You have 50 executor slots total. 20 are taken offline for patching. The spike hits with only 30 slots available.

Mitigation:

1. Never schedule maintenance during known peak hours (e.g., flash sale times).
2. Use rolling upgrades that only take 10% of nodes offline at any time.
3. Cross-AZ redundancy: If one AZ goes down for maintenance, the other AZs have enough capacity for 100% load.
4. Temporarily increase maxOffsetsPerTrigger to slow down consumption and prevent OOM on the reduced cluster, while ensuring Kafka's retention holds the data safely.

Sub-Scenario D: Multi-Topic Spike (3 Topics Spike Simultaneously)

Situation: You're reading from 3 Kafka topics (payments, refunds, notifications). All three spike at the same time.

Problem: A single maxOffsetsPerTrigger value applies to the TOTAL across all subscribed topics. If payments spike while refunds are normal, Spark may starve the refund topic.

Fix: Run separate streaming jobs per topic, each with independent rate limits and auto-scaling:

# DON'T: Single job reading all topics
spark.readStream.option("subscribe", "payments,refunds,notifications")

# DO: Separate jobs with independent configs
payments_job = spark.readStream.option("subscribe", "payments") \
    .option("maxOffsetsPerTrigger", 2000000).load()

refunds_job = spark.readStream.option("subscribe", "refunds") \
    .option("maxOffsetsPerTrigger", 500000).load()

Sub-Scenario E: The Spike Recurs Every Day at the Same Time

Situation: Every evening at 8 PM, traffic spikes 5x for 2 hours (dinner-time ordering pattern).

This is not a spike — it's a predictable pattern. The response should be proactive, not reactive:

1. Pre-scale: Use a cron job that adds executors at 7:45 PM (15 minutes before the spike).
2. Time-based maxOffsetsPerTrigger: Increase the rate limit during the 8-10 PM window.
3. Kafka partition rebalancing: Pre-warm partitions before the expected surge.
4. Cost optimization: Use spot/preemptible instances for the extra executors since the spike is time-bounded and you can tolerate a brief loss.

from datetime import datetime

current_hour = datetime.now().hour

if 19 <= current_hour <= 22:  # 7 PM to 10 PM
    max_offsets = 5000000    # Higher limit during peak
    max_executors = 80
else:
    max_offsets = 2000000    # Normal limit
    max_executors = 30