Real-Time Data Pipelines â Kafka & Flink Guide
In this tutorial, you'll learn about Real. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
A real-time data pipeline ingests, processes, and delivers data with latencies measured in seconds or milliseconds â using technologies like Kafka, Flink, and Debezium to process events as they happen.
What You'll Learn
This tutorial covers the complete real-time pipeline stack: building stream processors with Apache Kafka, writing streaming SQL with KSQL and Apache Flink SQL, capturing database changes with Debezium CDC, comparing Lambda vs Kappa architectures, and production deployment patterns.
Why It Matters
Batch pipelines have hours of latency. Real-time pipelines unlock immediate action â fraud detection, personalization, monitoring, and operational analytics. DodaTech's Durga Antivirus Pro processes threat intelligence streams in real time to block attacks within seconds of detection.
Real-World Use
Netflix uses Kafka Streams for real-time personalization of 200M+ subscriber recommendations. Uber processes 100B+ events daily through Flink for dispatch optimization. Shopify streams 1B+ events/minute through Kafka for inventory and checkout processing.
flowchart LR
subgraph Sources
A[Application
Events] --> D[Kafka]
B[Database CDC
Debezium] --> D
C[Log Files] --> D
end
subgraph Processing
D --> E[Kafka Streams
KStream/KTable]
D --> F[Flink SQL]
E --> G[State Store]
end
subgraph Sinks
G --> I[Real-time DB]
F --> J[Dashboard]
F --> K[Alert System]
end
Prerequisites: Python basics, SQL fundamentals. Understanding of ETL Pipelines and Data Engineering concepts helps.
Kafka Streams â Lightweight Stream Processing
Kafka Streams is a client library for building Stream Processing applications directly on top of Kafka, without requiring a separate processing cluster.
Core Abstractions
- KStream: An unbounded stream of records
- KTable: A changelog keyed by record key (table-like state)
- GlobalKTable: A fully replicated changelog for joins
class KafkaStreamsProcessor:
"""Simulate a Kafka Streams topology."""
def __init__(self, application_id):
self.app_id = application_id
self.state = {}
self.stores = {}
def kstream(self, topic):
return KStream(topic, self)
def ktable(self, topic):
return KTable(topic, self)
class KStream:
def __init__(self, topic, processor):
self.topic = topic
self.processor = processor
def filter(self, predicate):
ops = self.processor.state.setdefault(self.topic, [])
ops.append(("filter", predicate))
return self
def map_values(self, mapper):
ops = self.processor.state.setdefault(self.topic, [])
ops.append(("map_values", mapper))
return self
def process(self, records):
ops = self.processor.state.get(self.topic, [])
result = []
for record in records:
r = dict(record)
for op_type, *args in ops:
if op_type == "filter":
if not args[0](r):
r = None
break
elif op_type == "map_values":
r["value"] = args[0](r["value"])
if r:
result.append(r)
return result
class KTable:
def __init__(self, topic, processor):
self.topic = topic
self.processor = processor
self.table = {}
def load(self, records):
for r in records:
self.table[r["key"]] = r["value"]
def get(self, key):
return self.table.get(key)
proc = KafkaStreamsProcessor("clickstream_analytics")
stream = (proc.kstream("raw_clicks")
.filter(lambda r: r.get("value", {}).get("page") is not None)
.map_values(lambda v: {"page": v["page"], "category": v["page"].split("/")[1]}
if "/" in v.get("page", "") else v))
clicks = [
{"key": "u1", "value": {"page": "/python/tutorials"}},
{"key": "u2", "value": {"page": "/java/guide"}},
{"key": "u3", "value": {}},
{"key": "u1", "value": {"page": "/python/basics"}},
]
result = stream.process(clicks)
print("Filtered & transformed stream:")
for r in result:
print(f" {r['key']}: {r['value']}")
Expected output:
Filtered & transformed stream:
u1: {'page': '/python/tutorials', 'category': 'python'}
u2: {'page': '/java/guide', 'category': 'java'}
u1: {'page': '/python/basics', 'category': 'python'}
Stream-Table Joins
Enrich an order stream with a customer lookup table:
def stream_table_join():
proc = KafkaStreamsProcessor("order_enrichment")
customer_table = proc.ktable("customers")
customer_table.load([
{"key": "C001", "value": {"name": "Alice", "tier": "gold"}},
{"key": "C002", "value": {"name": "Bob", "tier": "silver"}},
])
order_stream = proc.kstream("orders")
orders = [
{"key": "ORD-001", "value": {"customer_id": "C001", "amount": 250}},
{"key": "ORD-002", "value": {"customer_id": "C002", "amount": 50}},
]
print("Enriched orders:")
for order in orders:
customer = customer_table.get(order["value"]["customer_id"])
enriched = {**order["value"], "customer_name": customer["name"], "tier": customer["tier"]}
print(f" {order['key']}: {enriched}")
stream_table_join()
Expected output:
Enriched orders:
ORD-001: {'customer_id': 'C001', 'amount': 250, 'customer_name': 'Alice', 'tier': 'gold'}
ORD-002: {'customer_id': 'C002', 'amount': 50, 'customer_name': 'Bob', 'tier': 'silver'}
KSQL â Streaming SQL for Kafka
KSQL provides a SQL interface for Stream Processing, letting you Express streaming operations as SQL statements without writing Java or Python.
-- Create a stream from a Kafka topic
CREATE STREAM clickstream (
user_id VARCHAR,
page VARCHAR,
event_time BIGINT
) WITH (
KAFKA_TOPIC = 'raw_clicks',
VALUE_FORMAT = 'JSON'
);
-- Continuous filtering query
CREATE STREAM mobile_clicks AS
SELECT user_id, page, event_time
FROM clickstream
WHERE page LIKE '/mobile/%'
EMIT CHANGES;
-- Windowed aggregation
CREATE TABLE hourly_page_views AS
SELECT page,
COUNT(*) AS views,
WINDOWSTART AS window_start
FROM clickstream
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY page
EMIT CHANGES;
class KSQLQueryEngine:
"""Simulate KSQL query execution."""
def __init__(self):
self.streams = {}
self.tables = {}
def create_stream(self, name, topic, schema):
self.streams[name] = {"topic": topic, "schema": schema, "events": [], "derived": []}
def ingest(self, stream_name, records):
if stream_name in self.streams:
self.streams[stream_name]["events"].extend(records)
def create_derived_stream(self, name, source, select_fields, where_clause=None):
source_stream = self.streams[source]
filtered = []
for event in source_stream["events"]:
if where_clause and not where_clause(event):
continue
filtered.append({f: event.get(f) for f in select_fields})
self.streams[name] = {"topic": name.lower(), "schema": {f: "unknown" for f in select_fields}, "events": filtered, "derived": [source]}
return filtered
ksql = KSQLQueryEngine()
ksql.create_stream("clickstream", "raw_clicks", {"user_id": "VARCHAR", "page": "VARCHAR"})
ksql.ingest("clickstream", [
{"user_id": "u1", "page": "/home"}, {"user_id": "u2", "page": "/mobile/offer"},
{"user_id": "u3", "page": "/mobile/checkout"}, {"user_id": "u1", "page": "/about"},
])
mobile = ksql.create_derived_stream("mobile_clicks", "clickstream", ["user_id", "page"],
where_clause=lambda e: "/mobile/" in e.get("page", ""))
print(f"Mobile clicks: {len(mobile)} events")
for m in mobile:
print(f" {m}")
Expected output:
Mobile clicks: 2 events
{'user_id': 'u2', 'page': '/mobile/offer'}
{'user_id': 'u3', 'page': '/mobile/checkout'}
Flink SQL â Unified Stream & Batch SQL
Apache Apache Flink SQL provides ANSI-compliant SQL with streaming semantics, supporting complex event-time processing, exactly-once state, and dynamic table materialization.
-- Flink SQL: streaming analytics
CREATE TABLE page_views (
user_id STRING,
page STRING,
view_time TIMESTAMP(3),
WATERMARK FOR view_time AS view_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'page_views',
'format' = 'json'
);
-- Windowed aggregation with event time
SELECT page, COUNT(*) AS views, TUMBLE_END(view_time, INTERVAL '10' MINUTE) AS window_end
FROM page_views
GROUP BY page, TUMBLE(view_time, INTERVAL '10' MINUTE);
class FlinkSQLExecutor:
def __init__(self):
self.tables = {}
def create_table(self, name, schema, watermark_column=None, watermark_delay=0):
self.tables[name] = {"schema": schema, "watermark": watermark_column, "delay": watermark_delay, "data": [], "watermark_value": 0}
def insert_into(self, table_name, records):
table = self.tables[table_name]
table["data"].extend(records)
if table["watermark"]:
max_ts = max(r.get(table["watermark"], 0) for r in records)
table["watermark_value"] = max_ts - table["delay"]
def select_tumble(self, source, group_by, agg, window_seconds=600, watermark_col="view_time"):
table = self.tables[source]
results = []
for record in table["data"]:
ts = record.get(watermark_col, 0)
if ts < table["watermark_value"]:
results.append({**record, "_window": ts // window_seconds})
groups = {}
for r in results:
key = r.get(group_by)
groups.setdefault(key, []).append(r)
final = []
for key, events in groups.items():
final.append({group_by: key, "count": len(events)})
return final
flink = FlinkSQLExecutor()
flink.create_table("page_views", {"user_id": "STRING", "page": "STRING", "view_time": "BIGINT"},
watermark_column="view_time", watermark_delay=5000)
flink.insert_into("page_views", [
{"user_id": "u1", "page": "/python", "view_time": 100000},
{"user_id": "u2", "page": "/java", "view_time": 150000},
{"user_id": "u1", "page": "/python", "view_time": 200000},
])
windows = flink.select_tumble("page_views", "page", "count", window_seconds=60)
print("Flink SQL tumbling window results:")
for w in sorted(windows, key=lambda x: x["count"], reverse=True):
print(f" {w['page']:<10} count={w['count']}")
Expected output:
Flink SQL tumbling window results:
python count=2
java count=1
CDC with Debezium
Debezium is an open-source platform for Change Data Capture (CDC), capturing row-level changes from databases and streaming them to Kafka.
class DebeziumCDC:
def __init__(self, connector_name, db_type="postgres"):
self.name = connector_name
self.db_type = db_type
self.change_log = []
def snapshot(self, tables):
for table, rows in tables.items():
for row in rows:
self.change_log.append({
"op": "r", "source": {"connector": self.name, "table": table},
"before": None, "after": row,
})
def capture_insert(self, table, row):
self.change_log.append({"op": "c", "source": {"connector": self.name, "table": table}, "before": None, "after": row})
def capture_update(self, table, before, after):
self.change_log.append({"op": "u", "source": {"connector": self.name, "table": table}, "before": before, "after": after})
def capture_delete(self, table, before):
self.change_log.append({"op": "d", "source": {"connector": self.name, "table": table}, "before": before, "after": None})
debezium = DebeziumCDC("orders-connector")
debezium.snapshot({"customers": [
{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"},
]})
debezium.capture_insert("orders", {"id": 101, "customer_id": 1, "total": 250.0})
debezium.capture_update("customers",
{"id": 1, "name": "Alice", "email": "alice@old.com"},
{"id": 1, "name": "Alice", "email": "alice@new.com"})
print(f"CDC events captured: {len(debezium.change_log)}")
for e in debezium.change_log:
print(f" op={e['op']} table={e['source']['table']} after={e['after']}")
Expected output:
CDC events captured: 4
op=r table=customers after={'id': 1, 'name': 'Alice'}
op=r table=customers after={'id': 2, 'name': 'Bob'}
op=c table=orders after={'id': 101, 'customer_id': 1, 'total': 250.0}
op=u table=customers after={'id': 1, 'name': 'Alice', 'email': 'alice@new.com'}
Lambda vs Kappa Architecture
| Aspect | Lambda | Kappa |
|---|---|---|
| Path | Batch + Streaming | Single streaming path |
| Code | Two codebases | One codebase |
| Storage | Lake (batch) + Speed layer | Append-only log (Kafka) |
| Reconciliation | Batch corrects stream | Reprocess from topic |
| Complexity | High â maintain both paths | Low â single pipeline |
| Latency | Batch: hours, Stream: seconds | Seconds to minutes |
Common Mistakes
1. Not Handling Schema Evolution
Data formats change over time. Use Avro or Protobuf with Schema Registry to handle backward/forward compatibility.
2. Ignoring Kafka Consumer Lag
Unchecked consumer lag indicates the pipeline can't keep up. Monitor with Burrow or Kafka's built-in consumer group commands.
3. No Idempotent Sinks
If your pipeline restarts, it may re-process events. Ensure sinks (DB, S3, API) are idempotent â writing the same record twice produces the same result.
4. Using Kafka Streams for Complex State
Kafka Streams is great for lightweight state (RocksDB), but for complex windowed joins or very large state, use Flink.
5. CDC Without Schema History
Debezium captures schema changes but without a schema history topic, table evolution breaks downstream consumers.
Practice Questions
1. What is the difference between KStream and KTable in Kafka Streams? KStream is an unbounded stream of records. KTable is a changelog keyed by record key, representing the latest state per key.
2. How does Debezium capture changes from a database? Debezium reads the database's Transaction log (WAL, binlog), converts row changes to structured events, and streams them to Kafka topics.
3. What is the key difference between Lambda and Kappa architectures? Lambda has separate batch and streaming paths (two codebases). Kappa uses a single streaming path with replay capability from the log.
4. What problem does watermarking solve in Flink SQL? Watermarks handle out-of-order events, telling Flink when to trigger window computations despite late-arriving data.
5. Challenge: Design a real-time pipeline for an e-commerce platform that needs to capture inventory changes from PostgreSQL, enrich with product data, and emit restocking alerts within 10 seconds.
Mini Project: Real-Time Pipeline Simulator
class RealTimePipeline:
def __init__(self):
self.cdc = DebeziumCDC("ecommerce-connector")
self.processor = KafkaStreamsProcessor("enrichment")
def run(self):
self.cdc.snapshot({"products": [
{"id": "P001", "name": "Laptop", "stock": 10},
{"id": "P002", "name": "Mouse", "stock": 50},
]})
self.cdc.capture_insert("orders", {"id": "ORD-101", "product_id": "P001", "qty": 1})
self.cdc.capture_update("products",
{"id": "P001", "name": "Laptop", "stock": 10},
{"id": "P001", "name": "Laptop", "stock": 9})
for msg in self.cdc.change_log:
print(f"[Sink] {msg['source']['table']}: op={msg['op']}")
RealTimePipeline().run()
FAQ
Related Concepts
What's Next
You now understand real-time pipelines with Kafka and Flink! Next, explore Data Pipeline Orchestration for managing complex workflows, and learn about Data Quality Monitoring for keeping your pipelines reliable in production.
- Practice daily â Build a Kafka Streams topology for clickstream data
- Build a project â Set up Debezium CDC for a PostgreSQL database
- Explore related topics â Check out Apache Spark Streaming for micro-batch processing
Remember: every expert was once a beginner. Keep coding!
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro