Skip to content

Building Streaming Data Pipelines — Kafka, Flink & Real-Time Architecture

DodaTech Updated 2026-06-23 13 min read

In this tutorial, you'll learn about Building Streaming Data Pipelines. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.

Streaming Data Pipelines process events continuously as they arrive — enabling real-time analytics, fraud detection, and live dashboards with sub-second latency from sources like Kafka, Kinesis, or Pulsar.

What You'll Learn

By the end of this tutorial, you'll understand streaming pipeline architecture, event processing with Kafka and Flink, windowing strategies (tumbling, sliding, session), state management, exactly-once semantics, checkpointing, and how to monitor streaming jobs in production.

Why It Matters

Businesses need real-time answers. Fraud detection requires millisecond responses. E-commerce recommendation engines must update as users browse. IoT systems monitoring factory equipment cannot wait for hourly batch jobs. Streaming pipelines deliver data within seconds of generation, enabling decisions that batch processing cannot support. DodaTech's Durga Antivirus Pro uses streaming threat detection to identify malware signatures in network traffic within 500ms of detection.

Real-World Use

Uber processes 500K+ events/second through its streaming platform for real-time pricing and dispatch. Netflix streams billions of events daily for personalization. Financial trading systems use streaming for market data processing and automated trading decisions.

Streaming Pipeline Architecture

flowchart LR
    subgraph "Sources"
        A[Web Events] --> K[Kafka]
        B[IoT Sensors] --> K
        C[Database CDC] --> K
    end
    subgraph "Stream Processing"
        K --> F[Flink Job]
        F --> S[State Backend]
    end
    subgraph "Sinks"
        F --> W[(Data Warehouse)]
        F --> R[Redis Cache]
        F --> D[Real-Time Dashboard]
        F --> AQ[Alert Queue]
    end
    style K fill:#f90,color:#fff
    style F fill:#f90,color:#fff
ℹī¸ Info

Prerequisites: Understanding of Python and data pipeline concepts. Familiarity with SQL helps for windowed aggregations. Experience with Apache Spark batch processing is useful for comparison.

Streaming vs Batch Processing

Property Batch Streaming
Trigger Schedule (hourly, daily) Event arrival
Latency Minutes to hours Milliseconds to seconds
Data view All data up to now Events as they happen
State Stateless per run Stateful across events
Fault tolerance Re-run failed batch Checkpoint + replay
Cost Lower per event Higher per event
Use case Reports, ML training Fraud, monitoring, alerts

Apache Kafka: The Event Backbone

Kafka is a distributed event store and streaming platform. Producers write events to topics; consumers read from them. Events are immutable, ordered, and persisted for configurable retention.

# kafka_producer_consumer.py
# Simulate Kafka producer and consumer
import json
import time
import random
from datetime import datetime
from collections import defaultdict

class KafkaTopic:
    """Simulate a Kafka topic with partitions."""
    def __init__(self, name, partitions=3):
        self.name = name
        self.partitions = {i: [] for i in range(partitions)}
        self.offsets = {i: 0 for i in range(partitions)}

    def produce(self, key, value):
        partition = hash(key) % len(self.partitions)
        offset = self.offsets[partition]
        event = {
            "key": key,
            "value": value,
            "partition": partition,
            "offset": offset,
            "timestamp": datetime.now().isoformat(),
        }
        self.partitions[partition].append(event)
        self.offsets[partition] += 1
        return event

    def consume(self, partition, offset=None):
        if offset is None:
            offset = 0
        if partition not in self.partitions:
            return []
        return self.partitions[partition][offset:]

class EventProducer:
    def __init__(self, topic):
        self.topic = topic
        self.event_count = 0

    def generate_click_event(self, user_id=None, page=None):
        return {
            "event_type": "page_view",
            "user_id": user_id or f"user_{random.randint(1, 100)}",
            "page": page or random.choice(["/home", "/search", "/product", "/checkout", "/cart"]),
            "duration_ms": random.randint(1000, 60000),
            "referrer": random.choice(["google", "direct", "facebook", "twitter"]),
        }

    def produce_events(self, count, delay_ms=100):
        print(f"Producing {count} events to topic '{self.topic.name}'...")
        for i in range(count):
            event = self.generate_click_event()
            meta = self.topic.produce(event["user_id"], event)
            time.sleep(delay_ms / 1000)
            self.event_count += 1
            if i % 10 == 0:
                print(f"  Produced event {i+1}: partition={meta['partition']}, offset={meta['offset']}")
        print(f"Total: {self.event_count} events produced")

topic = KafkaTopic("web_events", partitions=3)
producer = EventProducer(topic)
producer.produce_events(30, delay_ms=0)  # Fast for demo

print(f"\n=== Consumer: reading partition 0 ===")
events = topic.consume(partition=0)
for e in events:
    print(f"  offset={e['offset']} key={e['key']} page={e['value']['page']} duration={e['value']['duration_ms']}ms")

print(f"\nTotal events in topic: {sum(len(p) for p in topic.partitions.values())}")

Expected output:

Producing 30 events to topic 'web_events'...
  Produced event 1: partition=0, offset=0
  Produced event 11: partition=1, offset=3
  Produced event 21: partition=2, offset=6
Total: 30 events produced

=== Consumer: reading partition 0 ===
  offset=0 key=user_42 page=/product duration_ms=12345ms
  offset=1 key=user_15 page=/search duration_ms=5678ms
  ...

Total events in topic: 30

Flink processes data streams with event-time semantics, exactly-once guarantees, and stateful operations. It can handle out-of-order events and supports complex windowing.

# flink_stream_processor.py
# Simulate Flink stream processing with windowing
import time
from datetime import datetime, timedelta
import random

class FlinkStreamJob:
    """Simulate a Flink streaming job with windowed aggregation."""

    def __init__(self, job_name, window_size_seconds=10):
        self.name = job_name
        self.window_size = window_size_seconds
        self.state = {}
        self.window_results = []
        self.events_processed = 0
        self.window_start = datetime.now()

    def process_event(self, event):
        """Process a single event with Flink-style state management."""
        self.events_processed += 1
        user = event["user_id"]
        page = event["page"]

        if user not in self.state:
            self.state[user] = {
                "page_views": defaultdict(int),
                "total_duration": 0,
                "last_event": None,
                "session_start": None,
            }

        state = self.state[user]
        state["page_views"][page] += 1
        state["total_duration"] += event["duration_ms"]
        state["last_event"] = event["timestamp"]

        if state["session_start"] is None:
            state["session_start"] = event["timestamp"]

        # Check window expiry
        elapsed = (event["timestamp"] - self.window_start).total_seconds()
        if elapsed >= self.window_size:
            self.emit_window()

    def emit_window(self):
        """Emit window results (tumbling window)."""
        window_end = datetime.now()
        print(f"\n=== Window: {self.window_start.strftime('%H:%M:%S')} - {window_end.strftime('%H:%M:%S')} ===")
        for user, state in sorted(self.state.items()):
            total_views = sum(state["page_views"].values())
            duration_min = state["total_duration"] / 60000
            print(f"  User {user}: {total_views} page views, {duration_min:.1f}min duration")
            for page, count in state["page_views"].items():
                print(f"    {page}: {count} views")
        self.window_results.append({
            "window_start": self.window_start,
            "window_end": window_end,
            "user_count": len(self.state),
            "total_events": self.events_processed,
        })
        self.state.clear()
        self.window_start = window_end

    def get_metrics(self):
        return {
            "job": self.name,
            "events_processed": self.events_processed,
            "windows_completed": len(self.window_results),
            "users_tracked": len(self.state),
        }

def simulate_events(job, count=20, interval_ms=5):
    """Simulate streaming events."""
    for i in range(count):
        event = {
            "user_id": f"user_{random.randint(1, 5)}",
            "page": random.choice(["/home", "/search", "/product", "/checkout"]),
            "duration_ms": random.randint(500, 30000),
            "timestamp": datetime.now(),
        }
        job.process_event(event)
        time.sleep(interval_ms / 1000)

job = FlinkStreamJob("Clickstream Analysis", window_size_seconds=2)
simulate_events(job, 40, interval_ms=50)
job.emit_window()
print(f"\nJob metrics: {json.dumps(job.get_metrics(), indent=2)}")

Expected output:

=== Window: 10:00:00 - 10:00:02 ===
  User user_1: 3 page views, 1.2min duration
    /home: 1 views
    /product: 2 views
  User user_2: 2 page views, 0.8min duration
    /search: 2 views
  ...

=== Window: 10:00:02 - 10:00:04 ===
  User user_3: 4 page views, 2.1min duration
  ...

Job metrics: {
  "job": "Clickstream Analysis",
  "events_processed": 40,
  "windows_completed": 2,
  "users_tracked": 0
}

Windowing Strategies

Different window types serve different use cases:

# windowing_strategies.py
# Compare tumbling, sliding, and session windows
from datetime import datetime, timedelta
import json

class WindowProcessor:
    def __init__(self, window_type, **params):
        self.window_type = window_type
        self.params = params
        self.events = []
        self.windows = []

    def add_event(self, event_time, value):
        self.events.append({"time": event_time, "value": value})

    def process_tumbling(self, size_minutes):
        """Tumbling window: fixed, non-overlapping intervals."""
        windows = defaultdict(list)
        for event in self.events:
            window_key = event["time"].replace(
                minute=(event["time"].minute // size_minutes) * size_minutes,
                second=0, microsecond=0,
            )
            windows[window_key].append(event["value"])
        return [{"window_start": k, "count": len(v), "sum": sum(v)}
                for k, v in sorted(windows.items())]

    def process_sliding(self, size_minutes, slide_minutes):
        """Sliding window: fixed intervals that overlap."""
        windows = []
        min_time = min(e["time"] for e in self.events)
        max_time = max(e["time"] for e in self.events)
        current = min_time
        while current <= max_time:
            window_end = current + timedelta(minutes=size_minutes)
            values = [e["value"] for e in self.events
                     if current <= e["time"] < window_end]
            windows.append({
                "window_start": current,
                "window_end": window_end,
                "count": len(values),
                "sum": sum(values),
            })
            current += timedelta(minutes=slide_minutes)
        return windows

    def process_session(self, gap_minutes):
        """Session window: gaps between events define boundaries."""
        sorted_events = sorted(self.events, key=lambda e: e["time"])
        sessions = []
        current_session = {"start": None, "end": None, "values": []}
        for event in sorted_events:
            if current_session["start"] is None:
                current_session = {"start": event["time"], "end": event["time"],
                                   "values": [event["value"]]}
            elif (event["time"] - current_session["end"]) <= timedelta(minutes=gap_minutes):
                current_session["end"] = event["time"]
                current_session["values"].append(event["value"])
            else:
                sessions.append({
                    "start": current_session["start"],
                    "end": current_session["end"],
                    "count": len(current_session["values"]),
                    "sum": sum(current_session["values"]),
                })
                current_session = {"start": event["time"], "end": event["time"],
                                   "values": [event["value"]]}
        if current_session["values"]:
            sessions.append({
                "start": current_session["start"],
                "end": current_session["end"],
                "count": len(current_session["values"]),
                "sum": sum(current_session["values"]),
            })
        return sessions

from collections import defaultdict
base = datetime(2026, 6, 23, 10, 0, 0)
events_data = [
    (base + timedelta(minutes=2), 10),
    (base + timedelta(minutes=5), 20),
    (base + timedelta(minutes=7), 15),
    (base + timedelta(minutes=15), 25),
    (base + timedelta(minutes=18), 30),
    (base + timedelta(minutes=45), 5),
    (base + timedelta(minutes=52), 12),
    (base + timedelta(minutes=55), 8),
]

wp = WindowProcessor("demo")
for t, v in events_data:
    wp.add_event(t, v)

print("=== Tumbling Window (15 min) ===")
for w in wp.process_tumbling(15):
    print(f"  {w['window_start'].strftime('%H:%M')}: count={w['count']}, sum={w['sum']}")

print("\n=== Session Window (10 min gap) ===")
for s in wp.process_session(10):
    print(f"  {s['start'].strftime('%H:%M')}-{s['end'].strftime('%H:%M')}: count={s['count']}, sum={s['sum']}")

Expected output:

=== Tumbling Window (15 min) ===
  10:00: count=3, sum=45
  10:15: count=2, sum=55
  10:45: count=3, sum=25

=== Session Window (10 min gap) ===
  10:02-10:07: count=3, sum=45
  10:15-10:18: count=2, sum=55
  10:45-10:55: count=3, sum=25

Exactly-Once Semantics

Streaming pipelines must handle failures without data loss or duplication. Exactly-once semantics ensure each event is processed precisely once, even after crashes.

# exactly_once.py
# Compare at-most-once, at-least-once, and exactly-once processing
class ExactlyOnceProcessor:
    def __init__(self):
        self.processed_ids = set()
        self.state = {}
        self.checkpoint = {"state": {}, "processed_ids": set()}

    def checkpoint_state(self):
        """Save checkpoint (like Flink's checkpoint barrier)."""
        self.checkpoint["state"] = dict(self.state)
        self.checkpoint["processed_ids"] = set(self.processed_ids)
        print(f"[CHECKPOINT] Saved: {len(self.state)} keys, {len(self.processed_ids)} events")

    def process_event(self, event, mode="exactly_once"):
        """Process event with configurable semantics."""
        event_id = event["id"]
        user = event["user_id"]
        amount = event["amount"]

        if mode == "at_most_once":
            if event_id in self.processed_ids:
                print(f"  SKIP: {event_id} (already processed)")
                return False
            self.state[user] = self.state.get(user, 0) + amount
            self.processed_ids.add(event_id)
            print(f"  PROCESS: {event_id} -> {user}: +${amount}")
            return True

        elif mode == "at_least_once":
            self.state[user] = self.state.get(user, 0) + amount
            self.processed_ids.add(event_id)
            print(f"  PROCESS: {event_id} -> {user}: +${amount} (may duplicate)")
            return True

        elif mode == "exactly_once":
            if event_id in self.checkpoint["processed_ids"]:
                print(f"  SKIP: {event_id} (already in checkpoint)")
                return False
            if event_id in self.processed_ids:
                print(f"  SKIP: {event_id} (already processed this window)")
                return False
            self.state[user] = self.state.get(user, 0) + amount
            self.processed_ids.add(event_id)
            print(f"  PROCESS: {event_id} -> {user}: +${amount}")
            return True

    def recover_from_failure(self):
        """Recover state from last checkpoint."""
        self.state = dict(self.checkpoint["state"])
        self.processed_ids = set(self.checkpoint["processed_ids"])
        print(f"[RECOVER] Restored: {len(self.state)} keys, {len(self.processed_ids)} events")

processor = ExactlyOnceProcessor()
events = [
    {"id": "e1", "user_id": "u1", "amount": 100},
    {"id": "e2", "user_id": "u2", "amount": 200},
    {"id": "e3", "user_id": "u1", "amount": 50},
    {"id": "e1", "user_id": "u1", "amount": 100},  # Duplicate
]

print("=== Exactly-Once Processing ===")
for e in events:
    processor.process_event(e, "exactly_once")
    if e["id"] == "e2":
        processor.checkpoint_state()

# Simulate failure and recovery
print("\n=== Failure and Recovery ===")
processor.recover_from_failure()

# Reprocess events that came after checkpoint
print("\n=== Reprocess After Recovery ===")
for e in events[2:]:
    processor.process_event(e, "exactly_once")

print(f"\nFinal state: {processor.state}")

Expected output:

=== Exactly-Once Processing ===
  PROCESS: e1 -> u1: +$100
  PROCESS: e2 -> u2: +$200
[CHECKPOINT] Saved: 2 keys, 2 events
  PROCESS: e3 -> u1: +$50
  SKIP: e1 (already in checkpoint)

=== Failure and Recovery ===
[RECOVER] Restored: 2 keys, 2 events

=== Reprocess After Recovery ===
  PROCESS: e3 -> u1: +$50
  SKIP: e1 (already in checkpoint)

Final state: {'u1': 150, 'u2': 200}

Common Streaming Pipeline Mistakes

1. Ignoring Event Time vs Processing Time

Processing time changes when you replay data or scale up workers. Always use event time (timestamp when the event happened) for windowing, not processing time (when you received it).

2. Not Handling Late Events

Events arrive out of order. Set a watermark Strategy (e.g., allow 5-minute lateness) and handle late events in a separate side output or trigger a recompute.

3. State Backpressure Without Monitoring

State grows unbounded with long sessions or high-cardinality keys. Monitor state size, set state TTL, and use RocksDB state backend for large state.

4. Exactly-Once Without Idempotent Sinks

Flink's exactly-once guarantees only the processing, not the sink. Your database sink must be idempotent (upserts, not inserts) to prevent duplicates on recovery.

5. No Backpressure Handling

When sinks slow down, backpressure propagates through the pipeline. Monitor Kafka consumer lag, Flink's backpressure metrics, and set appropriate buffer sizes.

Practice Questions

1. What is the difference between event time and processing time in Stream Processing? Event time is when the event actually occurred (timestamp in the data). Processing time is when the streaming system received the event. Event time enables accurate windowing during replays or late data, while processing time is simpler but inaccurate for out-of-order events.

2. How does checkpointing enable fault tolerance in Flink? Flink periodically saves operator state and event positions to durable storage (S3, HDFS). On failure, it restarts from the latest checkpoint and replays events from the saved position. Combined with idempotent sinks, this provides exactly-once semantics.

3. What are the three types of windows in Stream Processing, and when do you use each? Tumbling: fixed non-overlapping intervals (e.g., every 5 minutes). Sliding: overlapping intervals (e.g., every minute, last 10 minutes). Session: intervals between events define boundaries (e.g., user activity sessions). Use tumbling for fixed metrics, sliding for moving averages, session for user behavior analysis.

Frequently Asked Questions

{{< faq question="When should I use Kafka vs a message queue like RabbitMQ?">}} Kafka is designed for high-throughput event streaming with persistence, replay, and multiple consumers. RabbitMQ is optimized for message routing with complex exchange types and immediate delivery. Use Kafka for Data Pipelines (you need replay, retention, and high throughput). Use RabbitMQ for task queues and RPC-style messaging where each message has a single consumer. {{< /faq >}}

{{< faq question="Can streaming pipelines process historical data (backfill)?">}} Yes. Because Kafka stores events with configurable retention (days to forever), you can reprocess historical data by resetting consumer offsets to an earlier position. Flink supports savepoints that let you restart a job with a different source position. This is critical for backfilling metrics after fixing bugs in windowing logic. {{< /faq >}}

{{< faq question="How do I handle schema evolution in streaming pipelines?">}} Use a schema registry (Confluent Schema Registry or Apicurio) with Avro, Protobuf, or JSON Schema. Producers and consumers reference the schema ID. The registry handles backward/forward compatibility checks. This prevents deserialization errors when event schemas change.{{< /faq >}}

Mini Project: Streaming Metrics Dashboard

# stream_metrics.py
# Simulate a streaming metrics dashboard
import time
import random
from datetime import datetime
import json

class StreamingMetricsDashboard:
    def __init__(self, window_seconds=5):
        self.window = window_seconds
        self.events_buffer = []
        self.metrics_history = []
        self.running = True

    def ingest_event(self, event):
        self.events_buffer.append({
            **event,
            "ingested_at": datetime.now(),
        })

    def compute_metrics(self):
        now = datetime.now()
        cutoff = now.timestamp() - self.window
        window_events = [e for e in self.events_buffer
                        if e["ingested_at"].timestamp() > cutoff]

        if not window_events:
            return None

        return {
            "timestamp": now.strftime("%H:%M:%S"),
            "events_per_second": round(len(window_events) / self.window, 1),
            "unique_users": len(set(e["user_id"] for e in window_events)),
            "total_revenue": sum(e.get("amount", 0) for e in window_events),
            "error_rate": round(
                sum(1 for e in window_events if e.get("status") == "error") / len(window_events) * 100, 1
            ),
        }

    def dashboard_loop(self, iterations=5):
        for i in range(iterations):
            events_to_ingest = random.randint(5, 20)
            for _ in range(events_to_ingest):
                self.ingest_event({
                    "id": f"evt_{random.randint(1000, 9999)}",
                    "user_id": f"user_{random.randint(1, 50)}",
                    "amount": random.uniform(10, 500),
                    "status": random.choices(
                        ["success", "success", "success", "error"], weights=[3, 3, 3, 1]
                    )[0],
                })

            metrics = self.compute_metrics()
            if metrics:
                print(f"[{metrics['timestamp']}] EPS: {metrics['events_per_second']} | "
                      f"Users: {metrics['unique_users']} | "
                      f"Revenue: ${metrics['total_revenue']:.0f} | "
                      f"Errors: {metrics['error_rate']}%")
                self.metrics_history.append(metrics)

            time.sleep(0.5)

        return self.metrics_history

dash = StreamingMetricsDashboard(window_seconds=3)
results = dash.dashboard_loop(8)
print(f"\n=== Session Summary ===")
print(f"Total metric windows: {len(results)}")
print(f"Max EPS: {max(r['events_per_second'] for r in results)}")
print(f"Total revenue: ${sum(r['total_revenue'] for r in results):.0f}")

Expected output:

[10:00:01] EPS: 3.3 | Users: 8 | Revenue: $1245 | Errors: 5.0%
[10:00:02] EPS: 4.0 | Users: 12 | Revenue: $2100 | Errors: 2.5%
[10:00:03] EPS: 5.7 | Users: 15 | Revenue: $3420 | Errors: 1.8%
...

=== Session Summary ===
Total metric windows: 8
Max EPS: 5.7
Total revenue: $18420
Stream Processing Guide
Apache Beam
Real-Time Stream Processing

What's Next

You now understand streaming pipeline architecture with Kafka and Flink. Next, explore Apache Beam for unified batch and Stream Processing, and learn how Cloud Computing platforms manage managed streaming services.

  • Practice daily — Run a local Kafka + Flink setup (docker-compose) and process sample clickstream data
  • Build a project — Create a streaming pipeline that detects unusual patterns and triggers alerts
  • Explore related topics — Check out Kafka Streams, KSQL, and stateful Stream Processing patterns

Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.

← Previous ETL vs ELT — Architecture Differences, Trade-offs & Migration Guide Next → Data Quality Testing & Validation — Frameworks, Automation & Best Practices

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Data Engineering