Learn real-time analytics architectures including stream processing, Lambda architecture, Kappa architecture, event-driven systems, and building real-time data pipelines with Apache Kafka and Apache Flink.
Learn real-time analytics architectures including stream processing, Lambda architecture, Kappa architecture, event-driven systems, and building real-time data pipelines with Apache Kafka and Apache Flink.
Real-time analytics enables organizations to process and analyze data as it arrives, providing immediate insights for time-sensitive decisions. Unlike traditional batch processing, real-time analytics systems process streaming data with low latency, enabling use cases like fraud detection, real-time recommendations, IoT monitoring, and operational dashboards.
This lesson explores the architectures, technologies, and patterns for building scalable real-time analytics systems.
Low Latency
: Process data within milliseconds to seconds of arrival
Continuous Processing
: Handle unbounded streams of data rather than fixed datasets
Event-Time Processing
: Process data based on when events occurred, not when they arrive
Fault Tolerance
: Ensure processing continues despite failures
Scalability
: Handle varying data volumes and velocity
Stateful Processing
: Maintain state across events for windowing and aggregations
Learn about Real-Time vs Batch Processing Comparison.
1
2┌─────────────────────────────────────────────────────────────────────┐
3│ Batch Processing vs Stream Processing │
4├─────────────────────────────────────────────────────────────────────┤
5│ │
6│ BATCH PROCESSING │
7│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
8│ │ Data │──▶│ Process │──▶│ Results │ │
9│ │ At Rest │ │ Complete │ │ Complete │ │
10│ │ (Hours) │ │ Dataset │ │ Dataset │ │
11│ └──────────┘ └──────────┘ └──────────┘ │
12│ │
13│ - High Latency (minutes to hours) │
14│ - Complete & Accurate Results │
15│ - Resource-efficient │
16│ - Simple to implement │
17│ │
18│ STREAM PROCESSING │
19│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
20│ │ Data │──▶│ Process │──▶│ Results │ │
21│ │ In │ │ Events │ │ Per │ │
22│ │ Motion │ │ As They │ │ Event │ │
23│ │ (ms) │ │ Arrive │ │ or Window│ │
24│ └──────────┘ └──────────┘ └──────────┘ │
25│ │
26│ - Low Latency (milliseconds to seconds) │
27│ - Approximate or incremental results │
28│ - Always-on processing │
29│ - Complex state management │
30│ │
31└─────────────────────────────────────────────────────────────────────┘
32Learn about Stream Processing Fundamentals.
Stream processing systems work with unbounded sequences of events. Each event has an event time (when it occurred) and a processing time (when the system processes it). Handling out-of-order events and late-arriving data is a fundamental challenge.
1# Event structure in a streaming system
2from dataclasses import dataclass
3from datetime import datetime
4from typing import Dict, Any
5
6@dataclass
7class Event:
8 """Represents a single event in a stream"""
9 event_id: str
10 event_type: str
11 event_time: datetime # When the event occurred
12 processing_time: datetime # When we received it
13 payload: Dict[str, Any]
14
15 def is_late(self, watermark: datetime) -> bool:
16 """Check if event arrived after the watermark"""
17 return self.event_time < watermark
18
19# Example: E-commerce purchase event
20purchase_event = Event(
21 event_id="evt_12345",
22 event_type="purchase",
23 event_time=datetime(2024, 1, 15, 10, 30, 0), # Event occurred at 10:30
24 processing_time=datetime(2024, 1, 15, 10, 30, 5), # Received 5 seconds later
25 payload={
26 "user_id": "user_789",
27 "product_id": "prod_456",
28 "amount": 99.99,
29 "currency": "USD"
30 }
31)
32Windows divide the infinite stream into finite chunks for aggregation. Different windowing strategies suit different use cases:
1
2┌─────────────────────────────────────────────────────────────────────┐
3│ Windowing Strategies │
4├─────────────────────────────────────────────────────────────────────┤
5│ │
6│ 1. TUMBLING WINDOWS (Fixed, Non-overlapping) │
7│ [────10min────][────10min────][────10min────] │
8│ Use: Hourly/daily aggregates, periodic reports │
9│ │
10│ 2. SLIDING WINDOWS (Fixed size, overlapping) │
11│ [────10min────] │
12│ [────10min────] │
13│ [────10min────] │
14│ Use: Moving averages, trending metrics │
15│ │
16│ 3. SESSION WINDOWS (Activity-based, variable size) │
17│ [──user_active──]......[──user_active──]...... │
18│ Use: User sessions, activity tracking │
19│ │
20│ 4. GLOBAL WINDOWS (All time, triggered by conditions) │
21│ [────────────────────all time───────────────────] │
22│ Use: Unbounded aggregations, custom triggers │
23│ │
24└─────────────────────────────────────────────────────────────────────┘
251# Apache Flink windowing examples
2from pyflink.datastream import StreamExecutionEnvironment
3from pyflink.datastream.window import TumblingEventTimeWindows, SlidingEventTimeWindows
4from pyflink.common import Time
5
6env = StreamExecutionEnvironment.get_execution_environment()
7
8# Tumbling window: Count purchases per 5-minute window
9tumbling_window = stream \\
10 .key_by(lambda event: event['product_id']) \\
11 .window(TumblingEventTimeWindows.of(Time.minutes(5))) \\
12 .aggregate(CountAggregateFunction())
13
14# Sliding window: Calculate 30-min average with 5-min slides
15sliding_window = stream \\
16 .key_by(lambda event: event['user_id']) \\
17 .window(SlidingEventTimeWindows.of(Time.minutes(30), Time.minutes(5))) \\
18 .aggregate(AverageAggregateFunction())
19
20# Session window: Track user sessions with 10-min timeout
21session_window = stream \\
22 .key_by(lambda event: event['user_id']) \\
23 .window(EventTimeSessionWindows.with_gap(Time.minutes(10))) \\
24 .aggregate(SessionAggregateFunction())
25Watermarks track event time progress in the stream, allowing the system to determine when a window can be closed. Late-arriving data can be handled through allowed lateness policies or side outputs.
1# Watermark strategies for handling out-of-order data
2from pyflink.datastream import WatermarkStrategy
3from datetime import timedelta
4
5# Periodic watermark with bounded out-of-orderness
6watermark_strategy = WatermarkStrategy \\
7 .for_bounded_out_of_orderness(timedelta(seconds=10)) \\
8 .with_timestamp_assigner(lambda event, ts: int(event['event_time'].timestamp() * 1000))
9
10stream_with_watermarks = raw_stream.assign_timestamps_and_watermarks(watermark_strategy)
11
12# Handling late data
13windowed_stream = stream_with_watermarks \\
14 .key_by(lambda event: event['user_id']) \\
15 .window(TumblingEventTimeWindows.of(Time.minutes(5))) \\
16 .allowed_lateness(Time.minutes(2)) # Accept late data up to 2 minutes \\
17 .side_output_late_data(late_data_output_tag) # Send very late data to side output
18Lambda Architecture combines batch and stream processing to provide both accuracy and low latency. It consists of three layers:
1
2┌─────────────────────────────────────────────────────────────────────┐
3│ Lambda Architecture │
4├─────────────────────────────────────────────────────────────────────┤
5│ │
6│ ┌──────────────────┐ │
7│ │ Data Sources │ │
8│ │ (Kafka, Kinesis)│ │
9│ └────────┬─────────┘ │
10│ │ │
11│ ┌────────────┴────────────┐ │
12│ │ │ │
13│ ▼ ▼ │
14│ ┌──────────────────┐ ┌──────────────────┐ │
15│ │ BATCH LAYER │ │ SPEED LAYER │ │
16│ │ (Spark, Hadoop) │ │ (Flink, Storm) │ │
17│ │ │ │ │ │
18│ │ - High accuracy │ │ - Low latency │ │
19│ │ - Immutable data │ │ - Approximate │ │
20│ │ - Recomputable │ │ - Incremental │ │
21│ └────────┬─────────┘ └────────┬─────────┘ │
22│ │ │ │
23│ │ Batch Views │ Real-time Views │
24│ │ │ │
25│ └────────┬───────────────┘ │
26│ ▼ │
27│ ┌──────────────────┐ │
28│ │ SERVING LAYER │ │
29│ │ (Druid, Cassandra) │
30│ │ │ │
31│ │ - Merge views │ │
32│ │ - Query interface│ │
33│ └────────┬─────────┘ │
34│ │ │
35│ ▼ │
36│ ┌──────────────────┐ │
37│ │ Query Results │ │
38│ └──────────────────┘ │
39│ │
40└─────────────────────────────────────────────────────────────────────┘
411# Batch layer: Compute accurate views from complete data
2from pyspark.sql import SparkSession
3from pyspark.sql.functions import col, window, count, sum, avg
4
5spark = SparkSession.builder.appName("BatchLayer").getOrCreate()
6
7# Read historical data (stored in data lake)
8historical_data = spark.read.parquet("s3://datalake/events/")
9
10# Compute batch views (e.g., daily user activity)
11batch_view = historical_data \\
12 .groupBy(window(col("event_time"), "1 day"), col("user_id")) \\
13 .agg(
14 count("*").alias("event_count"),
15 sum("amount").alias("total_amount"),
16 avg("session_duration").alias("avg_session_duration")
17 )
18
19# Write to serving layer
20batch_view.write \\
21 .format("parquet") \\
22 .mode("overwrite") \\
23 .partitionBy("window") \\
24 .save("s3://serving-layer/batch-views/user-activity/")
251# Speed layer: Process real-time events for immediate results
2from pyflink.datastream import StreamExecutionEnvironment
3from pyflink.datastream.window import TumblingEventTimeWindows
4from pyflink.common import Time
5
6env = StreamExecutionEnvironment.get_execution_environment()
7
8# Process streaming events (e.g., last hour of user activity)
9def process_realtime_events():
10 kafka_source = KafkaSource.builder() \\
11 .set_bootstrap_servers("localhost:9092") \\
12 .set_topics("user-events") \\
13 .set_value_only_deserializer(SimpleStringSchema()) \\
14 .build()
15
16 stream = env.from_source(kafka_source, WatermarkStrategy.no_watermarks(), "kafka")
17
18 # Compute real-time views
19 realtime_view = stream \\
20 .map(parse_event) \\
21 .key_by(lambda e: e['user_id']) \\
22 .window(TumblingEventTimeWindows.of(Time.minutes(5))) \\
23 .aggregate(UserActivityAggregator())
24
25 # Write to serving layer (Redis for fast queries)
26 realtime_view.add_sink(RedisSink())
27
28 env.execute("SpeedLayer")
291# Serving layer: Merge batch and real-time views
2import redis
3from cassandra.cluster import Cluster
4
5class ServingLayer:
6 def __init__(self):
7 self.redis_client = redis.Redis(host='localhost', port=6379)
8 self.cassandra_cluster = Cluster(['localhost'])
9 self.cassandra_session = self.cassandra_cluster.connect('analytics')
10
11 def query_user_activity(self, user_id: str, start_time: datetime, end_time: datetime):
12 """Merge batch and real-time views for complete picture"""
13
14 # Get batch view (complete, accurate historical data)
15 batch_query = """
16 SELECT window_start, event_count, total_amount
17 FROM batch_user_activity
18 WHERE user_id = %s AND window_start >= %s AND window_start < %s
19 """
20 batch_results = self.cassandra_session.execute(
21 batch_query, (user_id, start_time, end_time)
22 )
23
24 # Get real-time view (recent data, last few hours)
25 realtime_key = f"realtime:user:{user_id}"
26 realtime_results = self.redis_client.hgetall(realtime_key)
27
28 # Merge results
29 merged_results = self._merge_views(batch_results, realtime_results)
30 return merged_results
31
32 def _merge_views(self, batch_data, realtime_data):
33 """Combine batch and real-time results, preferring real-time for recent data"""
34 # Implementation details for merging logic
35 pass
36: Combines accuracy with low latency, fault-tolerant, handles reprocessing
: Complex to maintain (two code paths), requires merging logic, higher operational overhead
Kappa Architecture simplifies Lambda by eliminating the batch layer. All data processing happens in the stream processing layer, with the ability to reprocess data by replaying the event log.
1
2┌─────────────────────────────────────────────────────────────────────┐
3│ Kappa Architecture │
4├─────────────────────────────────────────────────────────────────────┤
5│ │
6│ ┌──────────────────┐ │
7│ │ Data Sources │ │
8│ │ (Kafka Topics) │ │
9│ └────────┬─────────┘ │
10│ │ │
11│ │ All data goes to log │
12│ ▼ │
13│ ┌──────────────────┐ │
14│ │ IMMUTABLE LOG │ │
15│ │ (Apache Kafka) │ │
16│ │ │ │
17│ │ - Durable │ │
18│ │ - Replayable │ │
19│ │ - Partitioned │ │
20│ └────────┬─────────┘ │
21│ │ │
22│ ┌────────────┴────────────┐ │
23│ │ │ │
24│ ▼ ▼ │
25│ ┌──────────────────┐ ┌──────────────────┐ │
26│ │ STREAM PROCESSOR │ │ STREAM PROCESSOR │ │
27│ │ (Version 1) │ │ (Version 2) │ │
28│ │ │ │ │ │
29│ │ - Current logic │ │ - New logic │ │
30│ │ - Real-time │ │ - Reprocessing │ │
31│ └────────┬─────────┘ └────────┬─────────┘ │
32│ │ │ │
33│ │ │ │
34│ ▼ ▼ │
35│ ┌──────────────────────────────────────────┐ │
36│ │ SERVING LAYER │ │
37│ │ (Cassandra, Elasticsearch, etc.) │ │
38│ └──────────────────┬───────────────────────┘ │
39│ │ │
40│ ▼ │
41│ ┌──────────────────┐ │
42│ │ Query Results │ │
43│ └──────────────────┘ │
44│ │
45│ Key Principle: Single code path for all processing │
46│ Reprocessing: Replay log with updated logic │
47│ │
48└─────────────────────────────────────────────────────────────────────┘
49Learn about Kappa Architecture Implementation.
1# Kappa architecture with Apache Flink
2from pyflink.datastream import StreamExecutionEnvironment
3from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkKafkaProducer
4
5class KappaProcessor:
6 def __init__(self, version: str):
7 self.env = StreamExecutionEnvironment.get_execution_environment()
8 self.version = version
9
10 def process_stream(self):
11 """Single processing pipeline for all data"""
12
13 # Read from Kafka (the immutable log)
14 kafka_consumer = FlinkKafkaConsumer(
15 topics='user-events',
16 deserialization_schema=JsonDeserializationSchema(),
17 properties={
18 'bootstrap.servers': 'localhost:9092',
19 'group.id': f'processor-{self.version}',
20 # For reprocessing, can start from beginning
21 'auto.offset.reset': 'earliest' if self.version == 'reprocess' else 'latest'
22 }
23 )
24
25 stream = self.env.add_source(kafka_consumer)
26
27 # Apply business logic (same code for real-time and reprocessing)
28 processed_stream = stream \\
29 .map(self.enrich_event) \\
30 .key_by(lambda e: e['user_id']) \\
31 .window(TumblingEventTimeWindows.of(Time.hours(1))) \\
32 .aggregate(UserMetricsAggregator())
33
34 # Write to serving layer
35 processed_stream.add_sink(self.create_cassandra_sink())
36
37 self.env.execute(f"KappaProcessor-{self.version}")
38
39 def enrich_event(self, event):
40 """Business logic for event processing"""
41 # Same logic applies whether processing real-time or reprocessing
42 enriched = {
43 **event,
44 'processed_version': self.version,
45 'enriched_at': datetime.now().isoformat()
46 }
47 return enriched
48
49# Running current version for real-time processing
50current_processor = KappaProcessor(version='v2')
51current_processor.process_stream()
52
53# Reprocessing with new logic (runs in parallel)
54reprocess_processor = KappaProcessor(version='v3-reprocess')
55reprocess_processor.process_stream()
56Learn about Kappa vs Lambda Architecture.
1
2┌─────────────────────────────────────────────────────────────────────┐
3│ Lambda vs Kappa Comparison │
4├─────────────────────────────────────────────────────────────────────┤
5│ │
6│ Aspect │ Lambda │ Kappa │
7│ ───────────────────────────────────────────────────────────────── │
8│ Code Paths │ Two (batch+stream) │ One (stream only) │
9│ Complexity │ Higher │ Lower │
10│ Accuracy │ Batch=exact │ Stream=approximate │
11│ Reprocessing │ Batch recompute │ Replay stream │
12│ Storage │ Data lake required │ Log is source of truth │
13│ Query Latency │ Merging overhead │ Direct queries │
14│ Operational Cost │ Higher │ Lower │
15│ Use Cases │ Complex analytics │ Event-driven apps │
16│ │ ML model training │ Real-time dashboards │
17│ │
18│ When to Use Lambda: │
19│ - Need absolute accuracy for historical data │
20│ - Complex batch algorithms (ML training) │
21│ - Large-scale reprocessing frequently needed │
22│ │
23│ When to Use Kappa: │
24│ - Stream processing covers all use cases │
25│ - Want to minimize operational complexity │
26│ - Event log can be retained long enough for reprocessing │
27│ │
28└─────────────────────────────────────────────────────────────────────┘
29Apache Kafka is the foundation of most real-time analytics architectures. It provides a distributed, fault-tolerant event streaming platform that can handle trillions of events per day.
Learn about Kafka Architecture.
1
2┌─────────────────────────────────────────────────────────────────────┐
3│ Apache Kafka Architecture │
4├─────────────────────────────────────────────────────────────────────┤
5│ │
6│ ┌─────────────────────────────────────────────────────────────┐ │
7│ │ PRODUCERS │ │
8│ │ (Web Servers, Microservices, IoT Devices, Databases) │ │
9│ └──────┬──────────────┬──────────────┬───────────────────────┘ │
10│ │ │ │ │
11│ ▼ ▼ ▼ │
12│ ┌─────────────────────────────────────────────────────────────┐ │
13│ │ KAFKA CLUSTER │ │
14│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │
15│ │ │ Broker 1 │ │ Broker 2 │ │ Broker 3 │ │ │
16│ │ │ │ │ │ │ │ │ │
17│ │ │ Topic: users │ │ Topic: users │ │ Topic: orders│ │ │
18│ │ │ Partition 0 │ │ Partition 1 │ │ Partition 0 │ │ │
19│ │ │ [──logs──] │ │ [──logs──] │ │ [──logs──] │ │ │
20│ │ │ Replica │ │ Leader │ │ Replica │ │ │
21│ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │
22│ │ │ │
23│ │ ┌──────────────────────────────────────────────────────┐ │ │
24│ │ │ ZooKeeper / KRaft │ │ │
25│ │ │ (Cluster coordination, leader election, metadata) │ │ │
26│ │ └──────────────────────────────────────────────────────┘ │ │
27│ └─────────────────────────────────────────────────────────────┘ │
28│ │ │ │ │
29│ ▼ ▼ ▼ │
30│ ┌─────────────────────────────────────────────────────────────┐ │
31│ │ CONSUMERS │ │
32│ │ (Stream Processors, Data Warehouses, Analytics Engines) │ │
33│ └─────────────────────────────────────────────────────────────┘ │
34│ │
35└─────────────────────────────────────────────────────────────────────┘
36Learn about Kafka Producer Example.
1# Producing events to Kafka
2from kafka import KafkaProducer
3import json
4from datetime import datetime
5
6class EventProducer:
7 def __init__(self, bootstrap_servers):
8 self.producer = KafkaProducer(
9 bootstrap_servers=bootstrap_servers,
10 value_serializer=lambda v: json.dumps(v).encode('utf-8'),
11 # Durability settings
12 acks='all', # Wait for all replicas to acknowledge
13 retries=3,
14 # Batching for performance
15 batch_size=16384,
16 linger_ms=10,
17 compression_type='snappy'
18 )
19
20 def send_user_event(self, user_id, event_type, data):
21 """Send user event to Kafka topic"""
22 event = {
23 'user_id': user_id,
24 'event_type': event_type,
25 'event_time': datetime.now().isoformat(),
26 'data': data
27 }
28
29 # Send with partition key (all events for same user go to same partition)
30 future = self.producer.send(
31 topic='user-events',
32 key=user_id.encode('utf-8'),
33 value=event
34 )
35
36 # Optional: Wait for acknowledgment
37 try:
38 record_metadata = future.get(timeout=10)
39 print(f"Sent to partition {record_metadata.partition} at offset {record_metadata.offset}")
40 except Exception as e:
41 print(f"Failed to send event: {e}")
42
43 def close(self):
44 self.producer.flush()
45 self.producer.close()
46
47# Usage
48producer = EventProducer(['localhost:9092'])
49producer.send_user_event('user_123', 'page_view', {'page': '/products', 'duration': 45})
50producer.send_user_event('user_123', 'purchase', {'product_id': 'prod_456', 'amount': 99.99})
51producer.close()
52Learn about Kafka Consumer Example.
1# Consuming events from Kafka
2from kafka import KafkaConsumer
3import json
4
5class EventConsumer:
6 def __init__(self, topics, group_id, bootstrap_servers):
7 self.consumer = KafkaConsumer(
8 *topics,
9 bootstrap_servers=bootstrap_servers,
10 group_id=group_id,
11 value_deserializer=lambda m: json.loads(m.decode('utf-8')),
12 # Consumer behavior
13 auto_offset_reset='earliest', # Start from beginning if no offset
14 enable_auto_commit=False, # Manual offset management for reliability
15 max_poll_records=500
16 )
17
18 def process_events(self):
19 """Process events in a loop"""
20 try:
21 for message in self.consumer:
22 try:
23 # Process the event
24 event = message.value
25 self.handle_event(event)
26
27 # Commit offset after successful processing
28 self.consumer.commit()
29
30 except Exception as e:
31 print(f"Error processing event: {e}")
32 # Could send to dead letter queue
33
34 except KeyboardInterrupt:
35 print("Shutting down consumer...")
36 finally:
37 self.consumer.close()
38
39 def handle_event(self, event):
40 """Business logic for event processing"""
41 event_type = event.get('event_type')
42
43 if event_type == 'purchase':
44 self.process_purchase(event)
45 elif event_type == 'page_view':
46 self.process_page_view(event)
47
48 def process_purchase(self, event):
49 print(f"Processing purchase: {event}")
50 # Update user metrics, trigger recommendations, etc.
51
52 def process_page_view(self, event):
53 print(f"Processing page view: {event}")
54 # Track user behavior, update analytics, etc.
55
56# Usage
57consumer = EventConsumer(
58 topics=['user-events'],
59 group_id='analytics-processor',
60 bootstrap_servers=['localhost:9092']
61)
62consumer.process_events()
63Learn about Kafka Streams for Stateful Processing.
1// Kafka Streams for complex stateful processing
2import org.apache.kafka.streams.KafkaStreams;
3import org.apache.kafka.streams.StreamsBuilder;
4import org.apache.kafka.streams.kstream.*;
5import java.time.Duration;
6
7public class UserAnalyticsStream {
8 public static void main(String[] args) {
9 StreamsBuilder builder = new StreamsBuilder();
10
11 // Read user events stream
12 KStream<String, UserEvent> events = builder.stream("user-events");
13
14 // Calculate real-time user metrics with windowing
15 KTable<Windowed<String>, UserMetrics> userMetrics = events
16 .groupByKey()
17 .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).advanceBy(Duration.ofMinutes(1)))
18 .aggregate(
19 UserMetrics::new, // Initializer
20 (userId, event, metrics) -> {
21 // Aggregation logic
22 metrics.incrementEventCount();
23 metrics.updateLastSeen(event.getTimestamp());
24 if (event.getType().equals("purchase")) {
25 metrics.addRevenue(event.getAmount());
26 }
27 return metrics;
28 },
29 Materialized.with(Serdes.String(), new UserMetricsSerde())
30 );
31
32 // Join streams: Enrich events with user profile
33 KTable<String, UserProfile> userProfiles = builder.table("user-profiles");
34
35 KStream<String, EnrichedEvent> enrichedEvents = events
36 .join(
37 userProfiles,
38 (event, profile) -> new EnrichedEvent(event, profile),
39 Joined.with(Serdes.String(), new EventSerde(), new ProfileSerde())
40 );
41
42 // Detect anomalies (e.g., sudden spike in purchases)
43 enrichedEvents
44 .filter((userId, event) -> isAnomalous(event))
45 .to("anomaly-alerts");
46
47 // Write aggregated metrics to output topic
48 userMetrics
49 .toStream()
50 .to("user-metrics");
51
52 // Start the streams application
53 KafkaStreams streams = new KafkaStreams(builder.build(), getStreamsConfig());
54 streams.start();
55
56 // Graceful shutdown
57 Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
58 }
59
60 private static boolean isAnomalous(EnrichedEvent event) {
61 // Anomaly detection logic
62 return event.getAmount() > event.getUserProfile().getAveragePurchase() * 5;
63 }
64}
65Apache Flink is a powerful distributed stream processing framework that provides exactly-once semantics, stateful computations, and advanced windowing capabilities.
Learn about Flink Processing Model.
1
2┌─────────────────────────────────────────────────────────────────────┐
3│ Apache Flink Architecture │
4├─────────────────────────────────────────────────────────────────────┤
5│ │
6│ ┌─────────────────────────────────────────────────────────────┐ │
7│ │ JOB MANAGER │ │
8│ │ - Scheduling │ │
9│ │ - Checkpointing coordination │ │
10│ │ - Recovery │ │
11│ └──────────────────────┬───────────────────────────────────────┘ │
12│ │ │
13│ ┌───────────────┼───────────────┐ │
14│ │ │ │ │
15│ ▼ ▼ ▼ │
16│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
17│ │ TASK │ │ TASK │ │ TASK │ │
18│ │ MANAGER 1 │ │ MANAGER 2 │ │ MANAGER 3 │ │
19│ │ │ │ │ │ │ │
20│ │ ┌────────┐ │ │ ┌────────┐ │ │ ┌────────┐ │ │
21│ │ │Task │ │ │ │Task │ │ │ │Task │ │ │
22│ │ │Slot 1 │ │ │ │Slot 1 │ │ │ │Slot 1 │ │ │
23│ │ └────────┘ │ │ └────────┘ │ │ └────────┘ │ │
24│ │ ┌────────┐ │ │ ┌────────┐ │ │ ┌────────┐ │ │
25│ │ │Task │ │ │ │Task │ │ │ │Task │ │ │
26│ │ │Slot 2 │ │ │ │Slot 2 │ │ │ │Slot 2 │ │ │
27│ │ └────────┘ │ │ └────────┘ │ │ └────────┘ │ │
28│ │ │ │ │ │ │ │
29│ │ Local │ │ Local │ │ Local │ │
30│ │ State │ │ State │ │ State │ │
31│ └─────────────┘ └─────────────┘ └─────────────┘ │
32│ │ │ │ │
33│ └───────────────┼───────────────┘ │
34│ ▼ │
35│ ┌────────────────────┐ │
36│ │ CHECKPOINTS │ │
37│ │ (Distributed │ │
38│ │ Snapshots) │ │
39│ │ - State backup │ │
40│ │ - Exactly-once │ │
41│ └────────────────────┘ │
42│ │
43└─────────────────────────────────────────────────────────────────────┘
44Learn about Flink DataStream API Example.
1# Complex event processing with Flink
2from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
3from pyflink.datastream.window import TumblingEventTimeWindows, SlidingEventTimeWindows
4from pyflink.datastream.functions import ProcessFunction, KeyedProcessFunction
5from pyflink.common import Time, Types
6from datetime import datetime, timedelta
7
8class RealTimeAnalyticsPipeline:
9 def __init__(self):
10 self.env = StreamExecutionEnvironment.get_execution_environment()
11 self.env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
12 self.env.enable_checkpointing(60000) # Checkpoint every minute
13
14 def create_pipeline(self):
15 # Source: Read from Kafka
16 kafka_source = self._create_kafka_source()
17 event_stream = self.env.from_source(kafka_source, watermark_strategy, "kafka")
18
19 # 1. Real-time session tracking
20 user_sessions = event_stream \\
21 .key_by(lambda e: e['user_id']) \\
22 .process(SessionTrackingFunction())
23
24 # 2. Aggregated metrics with tumbling windows
25 hourly_metrics = event_stream \\
26 .key_by(lambda e: e['product_id']) \\
27 .window(TumblingEventTimeWindows.of(Time.hours(1))) \\
28 .aggregate(ProductMetricsAggregator())
29
30 # 3. Sliding window for trend detection
31 trending_products = event_stream \\
32 .filter(lambda e: e['event_type'] == 'purchase') \\
33 .key_by(lambda e: e['product_id']) \\
34 .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(10))) \\
35 .aggregate(TrendingAggregator()) \\
36 .filter(lambda m: m['purchase_count'] > 100)
37
38 # 4. CEP for pattern detection (e.g., cart abandonment)
39 abandoned_carts = self._detect_cart_abandonment(event_stream)
40
41 # Sinks
42 user_sessions.add_sink(self._create_cassandra_sink("user_sessions"))
43 hourly_metrics.add_sink(self._create_cassandra_sink("product_metrics"))
44 trending_products.add_sink(self._create_kafka_sink("trending-products"))
45 abandoned_carts.add_sink(self._create_kafka_sink("abandoned-cart-alerts"))
46
47 self.env.execute("RealTimeAnalyticsPipeline")
48
49class SessionTrackingFunction(KeyedProcessFunction):
50 """Track user sessions with timeout-based termination"""
51
52 def process_element(self, event, ctx):
53 # Get current session state
54 session_state = ctx.get_state(ValueStateDescriptor("session", Types.STRING()))
55 current_session = session_state.value()
56
57 if current_session is None:
58 # Start new session
59 session_id = self._generate_session_id(event['user_id'], ctx.timestamp())
60 current_session = {
61 'session_id': session_id,
62 'user_id': event['user_id'],
63 'start_time': ctx.timestamp(),
64 'events': []
65 }
66
67 # Add event to session
68 current_session['events'].append(event)
69 session_state.update(current_session)
70
71 # Register timer to close session after 30 minutes of inactivity
72 timeout = ctx.timestamp() + 30 * 60 * 1000 # 30 minutes
73 ctx.timer_service().register_event_time_timer(timeout)
74
75 yield event
76
77 def on_timer(self, timestamp, ctx):
78 # Session timeout - close and emit session
79 session_state = ctx.get_state(ValueStateDescriptor("session", Types.STRING()))
80 session = session_state.value()
81
82 if session:
83 session['end_time'] = timestamp
84 session['duration'] = timestamp - session['start_time']
85 yield session
86 session_state.clear()
87
88class ProductMetricsAggregator(AggregateFunction):
89 """Aggregate product metrics over time windows"""
90
91 def create_accumulator(self):
92 return {
93 'view_count': 0,
94 'purchase_count': 0,
95 'total_revenue': 0.0,
96 'unique_users': set()
97 }
98
99 def add(self, event, accumulator):
100 if event['event_type'] == 'product_view':
101 accumulator['view_count'] += 1
102 elif event['event_type'] == 'purchase':
103 accumulator['purchase_count'] += 1
104 accumulator['total_revenue'] += event['amount']
105
106 accumulator['unique_users'].add(event['user_id'])
107 return accumulator
108
109 def get_result(self, accumulator):
110 return {
111 'view_count': accumulator['view_count'],
112 'purchase_count': accumulator['purchase_count'],
113 'total_revenue': accumulator['total_revenue'],
114 'unique_users': len(accumulator['unique_users']),
115 'conversion_rate': (accumulator['purchase_count'] / accumulator['view_count']
116 if accumulator['view_count'] > 0 else 0)
117 }
118
119 def merge(self, acc1, acc2):
120 return {
121 'view_count': acc1['view_count'] + acc2['view_count'],
122 'purchase_count': acc1['purchase_count'] + acc2['purchase_count'],
123 'total_revenue': acc1['total_revenue'] + acc2['total_revenue'],
124 'unique_users': acc1['unique_users'].union(acc2['unique_users'])
125 }
126Learn about Flink State Management and Checkpointing.
1# State management and fault tolerance in Flink
2from pyflink.datastream.state import ValueStateDescriptor, ListStateDescriptor, MapStateDescriptor
3
4class StatefulProcessor(KeyedProcessFunction):
5 """Example of using different state types in Flink"""
6
7 def open(self, runtime_context):
8 # Value state: Store single value per key
9 self.count_state = runtime_context.get_state(
10 ValueStateDescriptor("event_count", Types.LONG())
11 )
12
13 # List state: Store list of values per key
14 self.recent_events_state = runtime_context.get_list_state(
15 ListStateDescriptor("recent_events", Types.STRING())
16 )
17
18 # Map state: Store key-value pairs per key
19 self.feature_state = runtime_context.get_map_state(
20 MapStateDescriptor("features", Types.STRING(), Types.DOUBLE())
21 )
22
23 def process_element(self, event, ctx):
24 # Read and update value state
25 current_count = self.count_state.value() or 0
26 self.count_state.update(current_count + 1)
27
28 # Update list state (keep last 100 events)
29 self.recent_events_state.add(event)
30 if len(list(self.recent_events_state.get())) > 100:
31 events = list(self.recent_events_state.get())
32 self.recent_events_state.clear()
33 for e in events[-100:]:
34 self.recent_events_state.add(e)
35
36 # Update map state
37 self.feature_state.put("last_event_time", float(ctx.timestamp()))
38
39 yield {
40 'key': ctx.get_current_key(),
41 'event_count': current_count + 1,
42 'features': dict(self.feature_state.items())
43 }
44
45# Checkpoint configuration
46env = StreamExecutionEnvironment.get_execution_environment()
47
48# Enable checkpointing every 60 seconds
49env.enable_checkpointing(60000)
50
51# Checkpoint configuration
52checkpoint_config = env.get_checkpoint_config()
53checkpoint_config.set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)
54checkpoint_config.set_min_pause_between_checkpoints(30000) # 30 seconds
55checkpoint_config.set_checkpoint_timeout(600000) # 10 minutes
56checkpoint_config.set_max_concurrent_checkpoints(1)
57checkpoint_config.enable_externalized_checkpoints(
58 ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
59)
60
61# State backend configuration (RocksDB for large state)
62env.set_state_backend(RocksDBStateBackend("hdfs://namenode:9000/flink/checkpoints"))
63Learn about Real-World Real-Time Analytics Use Cases.
Learn about 1. Fraud Detection System.
1# Real-time fraud detection with Flink CEP (Complex Event Processing)
2from pyflink.datastream.functions import PatternProcessFunction
3from pyflink.cep import Pattern, CEP
4
5class FraudDetectionPipeline:
6 def create_fraud_detection_pattern(self, transaction_stream):
7 """Detect suspicious transaction patterns"""
8
9 # Pattern: Multiple high-value transactions in short time
10 suspicious_pattern = Pattern.begin("first_transaction") \\
11 .where(lambda txn: txn['amount'] > 1000) \\
12 .next("second_transaction") \\
13 .where(lambda txn: txn['amount'] > 1000) \\
14 .within(Time.minutes(5))
15
16 # Pattern: Transactions from unusual locations
17 location_pattern = Pattern.begin("home_location") \\
18 .where(lambda txn: txn['country'] == 'US') \\
19 .followed_by("foreign_location") \\
20 .where(lambda txn: txn['country'] != 'US') \\
21 .within(Time.hours(1))
22
23 # Apply patterns
24 suspicious_txns = CEP.pattern(
25 transaction_stream.key_by(lambda t: t['user_id']),
26 suspicious_pattern
27 ).process(SuspiciousTransactionDetector())
28
29 return suspicious_txns
30
31class SuspiciousTransactionDetector(PatternProcessFunction):
32 def process_match(self, match, ctx):
33 first = match['first_transaction'][0]
34 second = match['second_transaction'][0]
35
36 alert = {
37 'user_id': first['user_id'],
38 'alert_type': 'multiple_high_value_transactions',
39 'transactions': [first, second],
40 'risk_score': self.calculate_risk_score(first, second),
41 'timestamp': ctx.timestamp()
42 }
43
44 yield alert
45
46 def calculate_risk_score(self, txn1, txn2):
47 # ML-based risk scoring
48 score = 0.0
49 score += 0.3 if (txn2['amount'] / txn1['amount']) > 2 else 0.1
50 score += 0.4 if txn1['merchant_category'] != txn2['merchant_category'] else 0.1
51 return score
52Learn about 2. Real-Time Recommendation Engine.
1# Real-time personalization based on user behavior
2class RealtimeRecommendationEngine:
3 def __init__(self):
4 self.env = StreamExecutionEnvironment.get_execution_environment()
5
6 def create_recommendation_pipeline(self):
7 # Stream of user interactions
8 interaction_stream = self.env.from_source(kafka_source, watermark_strategy, "kafka")
9
10 # Calculate user preferences in real-time
11 user_preferences = interaction_stream \\
12 .key_by(lambda e: e['user_id']) \\
13 .process(UserPreferenceCalculator())
14
15 # Generate recommendations
16 recommendations = user_preferences \\
17 .flat_map(RecommendationGenerator())
18
19 # Write to low-latency store (Redis)
20 recommendations.add_sink(RedisSink())
21
22class UserPreferenceCalculator(KeyedProcessFunction):
23 """Calculate user preferences based on recent interactions"""
24
25 def open(self, runtime_context):
26 # State: Recent product views with decay
27 self.product_scores = runtime_context.get_map_state(
28 MapStateDescriptor("product_scores", Types.STRING(), Types.DOUBLE())
29 )
30
31 # State: Category preferences
32 self.category_scores = runtime_context.get_map_state(
33 MapStateDescriptor("category_scores", Types.STRING(), Types.DOUBLE())
34 )
35
36 def process_element(self, event, ctx):
37 event_type = event['event_type']
38
39 # Update scores based on event type
40 if event_type == 'product_view':
41 self._update_score(self.product_scores, event['product_id'], 1.0)
42 self._update_score(self.category_scores, event['category'], 0.5)
43 elif event_type == 'add_to_cart':
44 self._update_score(self.product_scores, event['product_id'], 3.0)
45 self._update_score(self.category_scores, event['category'], 1.5)
46 elif event_type == 'purchase':
47 self._update_score(self.product_scores, event['product_id'], 10.0)
48 self._update_score(self.category_scores, event['category'], 5.0)
49
50 # Emit updated preferences
51 yield {
52 'user_id': event['user_id'],
53 'product_preferences': dict(self.product_scores.items()),
54 'category_preferences': dict(self.category_scores.items()),
55 'timestamp': ctx.timestamp()
56 }
57
58 def _update_score(self, state, key, increment):
59 current_score = state.get(key) or 0.0
60 # Apply decay and add new score
61 decayed_score = current_score * 0.95
62 new_score = decayed_score + increment
63 state.put(key, new_score)
64Learn about 3. IoT Sensor Monitoring.
1# Real-time IoT sensor data processing
2class IoTMonitoringPipeline:
3 def create_monitoring_pipeline(self):
4 # Stream of sensor readings
5 sensor_stream = self.env.from_source(iot_source, watermark_strategy, "iot")
6
7 # Detect anomalies
8 anomalies = sensor_stream \\
9 .key_by(lambda s: s['sensor_id']) \\
10 .process(AnomalyDetector())
11
12 # Calculate aggregates per sensor
13 sensor_stats = sensor_stream \\
14 .key_by(lambda s: s['sensor_id']) \\
15 .window(TumblingEventTimeWindows.of(Time.minutes(1))) \\
16 .aggregate(SensorStatsAggregator())
17
18 # Alert on critical conditions
19 critical_alerts = anomalies \\
20 .filter(lambda a: a['severity'] == 'critical')
21
22 critical_alerts.add_sink(AlertingSink())
23 sensor_stats.add_sink(TimeseriesDBSink())
24
25class AnomalyDetector(KeyedProcessFunction):
26 """Detect anomalies using statistical methods and ML"""
27
28 def open(self, runtime_context):
29 # Maintain running statistics
30 self.stats_state = runtime_context.get_value_state(
31 ValueStateDescriptor("stats", Types.STRING())
32 )
33
34 # Load ML model for anomaly detection
35 self.model = self._load_anomaly_model()
36
37 def process_element(self, reading, ctx):
38 stats = self.stats_state.value() or {'mean': 0, 'std': 1, 'count': 0}
39
40 # Update running statistics
41 stats = self._update_stats(stats, reading['value'])
42 self.stats_state.update(stats)
43
44 # Check for anomalies
45 z_score = abs(reading['value'] - stats['mean']) / stats['std']
46
47 # Use ML model for complex anomaly detection
48 features = self._extract_features(reading, stats)
49 ml_score = self.model.predict(features)
50
51 if z_score > 3 or ml_score > 0.8:
52 yield {
53 'sensor_id': reading['sensor_id'],
54 'timestamp': reading['timestamp'],
55 'value': reading['value'],
56 'expected_range': (stats['mean'] - 2*stats['std'],
57 stats['mean'] + 2*stats['std']),
58 'severity': 'critical' if z_score > 4 else 'warning',
59 'ml_anomaly_score': ml_score
60 }
61Learn about Best Practices for Real-Time Analytics.
Partition appropriately
: Use meaningful keys that distribute load evenly
State management
: Keep state size manageable, use RocksDB for large state
Backpressure handling
: Design systems to handle varying load gracefully
Horizontal scaling
: Add more consumers/processors to handle increased throughput
Checkpointing
: Enable regular checkpoints for stateful processing
Exactly-once semantics
: Use idempotent operations or transactional sinks
Dead letter queues
: Handle poison messages gracefully
Monitoring and alerting
: Detect and respond to failures quickly
Learn about 3. Optimize Performance.
1# Performance optimization techniques
2class OptimizedStreamProcessor:
3 def configure_performance(self, env):
4 # 1. Tune parallelism
5 env.set_parallelism(24) # Match number of Kafka partitions
6
7 # 2. Configure memory
8 env.get_config().set_task_manager_memory("4096m")
9
10 # 3. Enable object reuse for better performance
11 env.get_config().enable_object_reuse()
12
13 # 4. Configure network buffers
14 env.get_config().set_number_of_network_buffers(8192)
15
16 # 5. Use efficient serialization
17 env.get_config().register_type_with_kryo_serializer(
18 CustomType, CustomTypeSerializer
19 )
20
21 def optimize_operators(self, stream):
22 # Use map instead of flatMap when possible
23 optimized = stream \\
24 .map(fast_transformation) \\ # Better than flatMap for 1:1
25 .filter(lambda x: x is not None) \\ # Early filtering reduces downstream load
26 .key_by(lambda x: x['key']) \\
27 .reduce(IncrementalAggregator()) # Better than aggregate for simple cases
28
29 return optimized
30Choose appropriate watermark strategy
: Balance latency vs completeness
Handle late data
: Use allowed lateness and side outputs
Monitor watermark lag
: Detect stuck watermarks and processing delays
Use event time for correctness
: Don't rely on processing time for business logic
Learn about 5. Monitor and Debug.
1# Monitoring real-time pipelines
2class MonitoredPipeline:
3 def add_monitoring(self, stream):
4 # Add metrics
5 stream \\
6 .map(ProcessingFunction()) \\
7 .map(lambda x: self._record_metrics(x))
8
9 # Add side outputs for debugging
10 main_output = OutputTag("main", Types.STRING())
11 error_output = OutputTag("errors", Types.STRING())
12 late_output = OutputTag("late", Types.STRING())
13
14 # Log samples for debugging
15 stream.filter(lambda x: random.random() < 0.01) \\
16 .map(lambda x: f"Sample: {x}") \\
17 .add_sink(LoggingSink())
18
19 def _record_metrics(self, event):
20 # Record processing metrics
21 metrics.increment_counter("events_processed")
22 metrics.record_latency("processing_time", time.time() - event['ingestion_time'])
23 return event
24Your progress is saved in this browser, on this device only.