Streaming Deep Dive

Kafka, Spark Structured Streaming, Flink and the streaming patterns that come up in nearly every senior data engineering loop.

Kafka Fundamentals

Core Concepts

Partitioning

Topic: orders (6 partitions)
┌────────────┬────────────┬────────────┐
│ Partition 0│ Partition 1│ Partition 2│
│ user_1     │ user_2     │ user_3     │
│ user_7     │ user_8     │ user_9     │
└────────────┴────────────┴────────────┘
┌────────────┬────────────┬────────────┐
│ Partition 3│ Partition 4│ Partition 5│
│ user_4     │ user_5     │ user_6     │
│ user_10    │ user_11    │ user_12    │
└────────────┴────────────┴────────────┘

Partition Key Choices:

Sizing Rules

Partitions = max(
    desired_throughput / throughput_per_partition,
    num_consumers_in_group
)

# Example: 100K msg/sec, each partition handles 10K/sec
# Partitions = max(100K/10K, 20 consumers) = max(10, 20) = 20

Retention

Delivery Semantics

At-Most-Once

• Fire and forget
• May lose messages
• Use: Metrics where loss is okay

At-Least-Once

• Retry until ack
• May duplicate messages
• Use: Most pipelines (with dedup)

Exactly-Once

• No loss, no duplicates
• Requires: Idempotent producers + transactional consumers
• Use: Financial, billing

How to Achieve Exactly-Once:

# Kafka producer
producer = KafkaProducer(
    enable_idempotence=True,  # Prevents duplicates on retry
    transactional_id="my-txn-id"  # Enables transactions
)

# Spark consumer
spark.readStream \
    .format("kafka") \
    .option("kafka.isolation.level", "read_committed")  # Only committed txns

Windowing

Tumbling Window

Fixed, non-overlapping intervals.

Events: [1, 2, 3, 4, 5, 6, 7, 8, 9]
Window: 3 events
Result: [1,2,3], [4,5,6], [7,8,9]

.groupBy(window("timestamp", "5 minutes"))

Sliding Window

Overlapping intervals.

Events: [1, 2, 3, 4, 5]
Window: 3, Slide: 1
Result: [1,2,3], [2,3,4], [3,4,5]

.groupBy(window("timestamp", "10 minutes", "2 minutes"))  # 10min window, 2min slide

Session Window

Gap-based, ends after inactivity.

Events: [1, 2, 3, --gap--, 7, 8, --gap--, 15]
Gap: 3
Result: [1,2,3], [7,8], [15]

.groupBy(session_window("timestamp", "30 minutes"))  # Spark 3.2+

Watermarks & Late Data

What is a Watermark?

A watermark says: “I believe I’ve seen all events with timestamp <= W”

df.withWatermark("event_time", "10 minutes")

This means: Allow events up to 10 minutes late. Later events are dropped.

Handling Late Data

# Late data to separate stream (Flink)
late_data = windowed_stream.getSideOutput(late_output_tag)

Spark Structured Streaming

Basic Pattern

# Read
df = spark.readStream \
    .format("kafka") \
    .option("subscribe", "topic") \
    .option("startingOffsets", "latest") \
    .load()

# Transform
result = df.select(from_json(col("value"), schema).alias("data")) \
    .select("data.*") \
    .withWatermark("timestamp", "5 minutes") \
    .groupBy(window("timestamp", "1 minute")) \
    .count()

# Write
query = result.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/checkpoint") \
    .trigger(processingTime="30 seconds") \
    .start()

Output Modes

Triggers

Flink vs Spark Streaming

When to Use Flink

• Sub-second latency required
• Complex event processing (fraud patterns)
• Large state (>100GB)

When to Use Spark Streaming

• Team knows Spark
• Delta Lake integration
• Batch + streaming on same code
• Latency >10s is acceptable

State Management

Stateless vs Stateful

State Backends

State Cleanup

# Spark: State expires with watermark
df.withWatermark("timestamp", "1 hour")  # State cleared after 1 hour

# Flink: Explicit TTL
descriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.hours(1)).build())

Checkpointing

Periodic snapshots of state for fault tolerance.

# Spark
.option("checkpointLocation", "/checkpoints/my_job")

# Flink
env.enableCheckpointing(60000)  # Every 60 seconds
env.getCheckpointConfig().setCheckpointStorage("s3://bucket/checkpoints")

Recovery

On failure, restart from last checkpoint. Reprocess events since checkpoint.

Checkpoint vs Savepoint (Flink)

• Checkpoint: Automatic, for recovery
• Savepoint: Manual, for upgrades/migrations

Common Patterns

Deduplication

# Spark: Dedupe within watermark window
df.withWatermark("timestamp", "10 minutes") \
  .dropDuplicates(["event_id", "timestamp"])

# Alternative: Use Delta MERGE with event_id as key

Stream-Stream Join

# Join two streams within time range
left.withWatermark("time", "10 minutes") \
    .join(
        right.withWatermark("time", "10 minutes"),
        expr("left.key = right.key AND left.time BETWEEN right.time - interval 5 minutes AND right.time + interval 5 minutes")
    )

Stream-Static Join (Enrichment)

# Join streaming events with static dimension
events.join(broadcast(dim_product), "product_id")

References

• Kafka Documentation
• Spark Structured Streaming Guide
• Flink Documentation
• Streaming Systems Book

Stream-stream join: how state is bounded

sequenceDiagram
    participant I as impressions stream
    participant J as join operator (state store)
    participant C as clicks stream
    I->>J: impression ad=7 t=10:00 (buffered)
    C->>J: click ad=7 t=10:03
    J-->>J: match within 0-15 min interval
    Note over J: emits joined row
    Note over J: watermark passes 10:15 + delay, so impression state is evicted

imps   = impressions.withWatermark("imp_ts", "10 minutes")
clicks = clicks.withWatermark("click_ts", "20 minutes")
joined = imps.join(
    clicks,
    expr("""click_ad_id = imp_ad_id AND
            click_ts >= imp_ts AND
            click_ts <= imp_ts + interval 15 minutes"""),
    "leftOuter")          # outer joins REQUIRE watermarks + time bounds

Without watermarks and a time-range condition, the engine must keep both sides forever, and state grows without bound.

Interview questions