🧠 Flink Fundamentals

🎯 What is Apache Flink?

Apache Flink is a distributed stream processing framework for stateful computations over unbounded (stream) and bounded (batch) data. It provides true stream processing with low latency, exactly-once semantics, and event-time processing.

Why Flink Matters

Streaming-first: True stream processing. Stateful: Manage state at scale. Event-time: Handle out-of-order data. Exactly-once: Guaranteed processing. Low latency: Milliseconds. Unified: Batch + stream.

📌 Streaming-First📌 Stateful📌 Event-Time📌 Exactly-Once
# Apache Flink — the big picture # Flink is a distributed stream processing framework. # CONCEPTUAL OVERVIEW print(""" === Apache Flink Overview === Flink is: - Stream processing framework - Stateful computations - Event-time processing - Exactly-once semantics - Low latency (ms) - Unified batch + stream Key concepts: - DataStream: Unbounded stream - DataSet: Bounded batch (legacy) - State: Managed state - Time: Event-time, processing-time - Watermark: Event-time progress - Window: Group events - Checkpoint: Fault tolerance - Savepoint: Manual checkpoint Use cases: - Real-time analytics - Event-driven applications - Streaming ETL - Fraud detection - IoT analytics - Log processing """) # WORD COUNT IN FLINK (Python) print("=== Word Count in Flink (PyFlink) ===") print(""" from pyflink.datastream import StreamExecutionEnvironment from pyflink.datastream.functions import MapFunction, FlatMapFunction # Create environment env = StreamExecutionEnvironment.get_execution_environment() # Read from socket stream = env.socket_text_stream("localhost", 9999) # Word count counts = stream \\ .flat_map(lambda line: line.split()) \\ .map(lambda word: (word.lower(), 1)) \\ .key_by(lambda x: x[0]) \\ .reduce(lambda a, b: (a[0], a[1] + b[1])) # Print counts.print() # Execute env.execute("Word Count") """) print("\n=== Word Count in Flink (Java) ===") print(""" StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream text = env.socketTextStream("localhost", 9999); DataStream> counts = text .flatMap(new Tokenizer()) .keyBy(value -> value.f0) .sum(1); counts.print(); env.execute("Word Count"); // Tokenizer public static class Tokenizer implements FlatMapFunction> { @Override public void flatMap(String value, Collector> out) { for (String word : value.split("\\\\s")) { out.collect(new Tuple2<>(word.toLowerCase(), 1)); } } } """) # KEY PROPERTIES print(""" === Key Properties of Flink === 1. TRUE STREAMING - Event-at-a-time processing - Not micro-batch - Low latency (ms) 2. STATEFUL - Managed state - Keyed state - Operator state - Fault-tolerant 3. EVENT-TIME - Process by event time - Handle out-of-order - Watermarks 4. EXACTLY-ONCE - Checkpointing - Two-phase commit - No duplicates, no loss 5. WINDOWING - Tumbling - Sliding - Session - Custom 6. UNIFIED - Batch as bounded stream - Same API - Same runtime 7. SCALABLE - Horizontal scaling - Thousands of nodes - Millions of events/sec """) # REAL-WORLD SCALE print("=== Real-World Flink Deployments ===") deployments = [ ('Alibaba', 'Billions of events/day', 'Singles Day'), ('Uber', 'Trillions of events/day', 'Real-time analytics'), ('Netflix', 'Petabytes/day', 'Streaming ETL'), ('Capital One', 'Millions of events/sec', 'Fraud detection'), ('Bouygues Telecom', 'Billions of events/day', 'Network monitoring'), ('King (Candy Crush)', 'Millions of events/sec', 'Game analytics'), ] print(f"{'Company':<20} {'Scale':<30} {'Use Case':<25}") print("-" * 75) for company, scale, use_case in deployments: print(f"{company:<20} {scale:<30} {use_case:<25}")

🏗️ Architecture

🎯 Flink Architecture

Flink's architecture consists of a JobManager (master) and TaskManagers (workers). Jobs are submitted as dataflows and executed in parallel.

Architecture Components

JobManager: Master, coordination. TaskManager: Workers, execution. JobGraph: Logical plan. ExecutionGraph: Physical plan. Task: Unit of work. Slot: Resource unit.

🏗️ JobManager🏗️ TaskManager🏗️ JobGraph🏗️ Slots
# Flink Architecture print("=== Flink Architecture Diagram ===") print(""" ┌─────────────────────────────────────────┐ │ Client │ │ (Job Submission) │ └─────────────────┬───────────────────────┘ │ ▼ ┌─────────────────────────────────────────┐ │ JobManager │ │ (Master Node) │ │ │ │ - Job scheduling │ │ - Checkpoint coordination │ │ - Recovery │ │ - Resource management │ └─────────────────┬───────────────────────┘ │ ┌───────────┼───────────┐ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │TaskMgr 1│ │TaskMgr 2│ │TaskMgr 3│ │ │ │ │ │ │ │ ┌─────┐ │ │ ┌─────┐ │ │ ┌─────┐ │ │ │Slot │ │ │ │Slot │ │ │ │Slot │ │ │ │Task │ │ │ │Task │ │ │ │Task │ │ │ └─────┘ │ │ └─────┘ │ │ └─────┘ │ │ ┌─────┐ │ │ ┌─────┐ │ │ ┌─────┐ │ │ │Slot │ │ │ │Slot │ │ │ │Slot │ │ │ │Task │ │ │ │Task │ │ │ │Task │ │ │ └─────┘ │ │ └─────┘ │ │ └─────┘ │ └─────────┘ └─────────┘ └─────────┘ """) # COMPONENTS print("\n=== Flink Components ===") print(""" 1. JOBMANAGER - Master node - Job scheduling - Checkpoint coordination - Recovery - Resource management Components: - ResourceManager - Dispatcher - JobMaster 2. TASKMANAGER - Worker node - Task execution - State storage - Network communication - Multiple slots 3. SLOT - Resource unit - Memory, CPU - One task per slot - Multiple slots per TM 4. JOBGRAPH - Logical plan - Operators - Edges - Parallelism 5. EXECUTIONGRAPH - Physical plan - Tasks - Partitions - Parallel instances """) # EXECUTION MODES print("\n=== Execution Modes ===") print(""" 1. LOCAL - Single JVM - Testing - IDE development 2. STANDALONE - Flink cluster - Manual scaling - Simple 3. YARN - Hadoop integration - Resource sharing - Dynamic scaling 4. KUBERNETES - Container-native - Cloud-native - Auto-scaling 5. MESOS - Legacy - Not recommended """) # DATAFLOW print("\n=== Dataflow Execution ===") print(""" Source → Transformation → Sink Example: Kafka Source → Map → KeyBy → Window → Reduce → Kafka Sink Parallelism: - Each operator has parallelism - Data partitioned across instances - KeyBy groups by key Operator chaining: - Fuse operators - Reduce network overhead - Same thread """) # IMPLEMENTATION print("\n=== Simulating Flink Architecture ===") import time import threading from collections import defaultdict import queue class TaskManager: """Simulate Flink TaskManager.""" def __init__(self, tm_id, num_slots=4): self.id = tm_id self.num_slots = num_slots self.slots = [None] * num_slots # Slot -> task self.tasks = {} def allocate_slot(self, task_id): """Allocate a slot for a task.""" for i, slot in enumerate(self.slots): if slot is None: self.slots[i] = task_id self.tasks[task_id] = {'slot': i, 'status': 'running'} return i return None def release_slot(self, task_id): """Release a slot.""" if task_id in self.tasks: slot = self.tasks[task_id]['slot'] self.slots[slot] = None del self.tasks[task_id] return True return False def get_status(self): used_slots = sum(1 for s in self.slots if s is not None) return { 'id': self.id, 'slots': self.num_slots, 'used_slots': used_slots, 'available_slots': self.num_slots - used_slots, 'tasks': len(self.tasks), } class JobManager: """Simulate Flink JobManager.""" def __init__(self): self.task_managers = {} self.jobs = {} self.next_job_id = 1 def register_task_manager(self, tm): """Register a TaskManager.""" self.task_managers[tm.id] = tm print(f"JobManager: Registered TaskManager {tm.id} ({tm.num_slots} slots)") def submit_job(self, name, operators, parallelism): """Submit a job.""" job_id = f"job_{self.next_job_id}" self.next_job_id += 1 print(f"\nJobManager: Submitting job '{name}' (id={job_id})") print(f" Operators: {len(operators)}") print(f" Parallelism: {parallelism}") # Create tasks tasks = [] for op in operators: for i in range(parallelism): task_id = f"{job_id}_{op}_{i}" tasks.append({ 'id': task_id, 'operator': op, 'parallelism': i, }) # Allocate slots allocated = [] for task in tasks: for tm in self.task_managers.values(): slot = tm.allocate_slot(task['id']) if slot is not None: task['tm'] = tm.id task['slot'] = slot allocated.append(task) break self.jobs[job_id] = { 'id': job_id, 'name': name, 'tasks': allocated, 'status': 'running', } print(f" Tasks allocated: {len(allocated)}/{len(tasks)}") return job_id def get_status(self): """Get cluster status.""" return { 'task_managers': len(self.task_managers), 'total_slots': sum(tm.num_slots for tm in self.task_managers.values()), 'used_slots': sum(sum(1 for s in tm.slots if s is not None) for tm in self.task_managers.values()), 'jobs': len(self.jobs), } # Demo print("=== Flink Cluster Simulation ===") # Create JobManager jm = JobManager() # Create TaskManagers for i in range(1, 4): tm = TaskManager(f"tm_{i}", num_slots=4) jm.register_task_manager(tm) # Show cluster status status = jm.get_status() print(f"\nCluster Status:") print(f" TaskManagers: {status['task_managers']}") print(f" Total slots: {status['total_slots']}") print(f" Used slots: {status['used_slots']}") print(f" Available: {status['total_slots'] - status['used_slots']}") # Submit job job_id = jm.submit_job( 'word_count', ['source', 'flatMap', 'keyBy', 'reduce', 'sink'], parallelism=2 ) # Show job status job = jm.jobs[job_id] print(f"\nJob {job_id} tasks:") for task in job['tasks'][:5]: print(f" {task['id']} → {task['tm']} slot {task['slot']}") print(f" ... and {len(job['tasks']) - 5} more") # Show cluster status status = jm.get_status() print(f"\nCluster Status After Job:") print(f" Used slots: {status['used_slots']}/{status['total_slots']}") # ARCHITECTURE SUMMARY print("\n=== Architecture Summary ===") print(f"{'Component':<25} {'Role':<40}") print("-" * 65) print(f"{'JobManager':<25} {'Master, coordination':<40}") print(f"{'TaskManager':<25} {'Workers, execution':<40}") print(f"{'Slot':<25} {'Resource unit':<40}") print(f"{'JobGraph':<25} {'Logical plan':<40}") print(f"{'ExecutionGraph':<25} {'Physical plan':<40}") print(f"{'Task':<25} {'Unit of work':<40}")

📊 Dataflow Model

🎯 Streams and Transformations

Flink's dataflow model represents computations as a directed graph of operators connected by data streams. Data flows from sources through transformations to sinks.

Dataflow Model

Source: Data input. Transformation: Process data. Sink: Data output. DataStream: Unbounded. DataSet: Bounded. DAG: Dataflow graph.

📊 Source📊 Transformation📊 Sink📊 DAG
# Flink Dataflow Model print("=== Dataflow Overview ===") print(""" Dataflow: - Directed Acyclic Graph (DAG) - Operators = nodes - Streams = edges - Data flows through Structure: Source → Transformation → Transformation → Sink """) print("\n=== DataStream API ===") print(""" DataStream: - Unbounded stream of events - Immutable - Can be transformed - Typed Operations: - map, flatMap, filter - keyBy, reduce, aggregate - window, join, connect - union, split, iterate Example: DataStream lines = env.addSource(kafkaSource); DataStream> counts = lines .flatMap(new Tokenizer()) .keyBy(value -> value.f0) .sum(1); counts.addSink(kafkaSink); """) print("\n=== Operators ===") print(""" 1. SOURCE OPERATORS - Kafka - File - Socket - Collection - Custom 2. TRANSFORMATION OPERATORS - map: One-to-one - flatMap: One-to-many - filter: Subset - keyBy: Partition by key - reduce: Aggregate by key - window: Group by time - join: Combine streams - connect: Mix streams 3. SINK OPERATORS - Kafka - File - Database - Elasticsearch - Custom """) print("\n=== Parallelism ===") print(""" Parallelism: - Number of parallel instances - Each operator can have different parallelism - Data partitioned across instances Example: Source (p=2) → Map (p=4) → KeyBy → Reduce (p=4) → Sink (p=2) KeyBy: - Repartitions by key - Same key → same instance - Enables keyed state """) # IMPLEMENTATION print("\n=== Simulating Dataflow ===") class DataStream: """Simulate Flink DataStream.""" def __init__(self, data=None, operations=None): self.data = data or [] self.operations = operations or [] def map(self, func): """Map operation.""" new_ops = self.operations + [f"map({func})"] new_data = [func(x) for x in self.data] return DataStream(new_data, new_ops) def flatMap(self, func): """FlatMap operation.""" new_ops = self.operations + [f"flatMap({func})"] new_data = [] for x in self.data: new_data.extend(func(x)) return DataStream(new_data, new_ops) def filter(self, func): """Filter operation.""" new_ops = self.operations + [f"filter({func})"] new_data = [x for x in self.data if func(x)] return DataStream(new_data, new_ops) def keyBy(self, key_func): """KeyBy operation.""" new_ops = self.operations + [f"keyBy({key_func})"] return KeyedStream(self.data, key_func, new_ops) def print(self): """Print operation.""" print(f" Operations: {' → '.join(self.operations)}") for item in self.data[:10]: print(f" {item}") if len(self.data) > 10: print(f" ... and {len(self.data) - 10} more") class KeyedStream: """Simulate KeyedStream.""" def __init__(self, data, key_func, operations): self.data = data self.key_func = key_func self.operations = operations def reduce(self, reduce_func): """Reduce operation.""" new_ops = self.operations + [f"reduce({reduce_func})"] # Group by key grouped = defaultdict(list) for item in self.data: grouped[self.key_func(item)].append(item) # Reduce result = [] for key, items in grouped.items(): acc = items[0] for item in items[1:]: acc = reduce_func(acc, item) result.append(acc) return DataStream(result, new_ops) def sum(self, position): """Sum operation.""" def sum_reduce(a, b): if isinstance(a, tuple): result = list(a) result[position] = a[position] + b[position] return tuple(result) return a + b return self.reduce(sum_reduce) # Demo print("=== Dataflow Demo ===") # Create stream data = [ "hello world", "hello flink", "flink streaming", "hello world", ] print("1. Source:") stream = DataStream(data) for item in stream.data: print(f" {item}") print("\n2. FlatMap (tokenize):") words = stream.flatMap(lambda line: line.split()) for item in words.data: print(f" {item}") print("\n3. Map (word, 1):") pairs = words.map(lambda word: (word.lower(), 1)) for item in pairs.data: print(f" {item}") print("\n4. KeyBy + Sum:") counts = pairs.keyBy(lambda x: x[0]).sum(1) for item in counts.data: print(f" {item}") print("\n5. Pipeline:") counts.print() # DATAFLOW SUMMARY print("\n=== Dataflow Summary ===") print(f"{'Component':<25} {'Description':<40}") print("-" * 65) print(f"{'Source':<25} {'Data input':<40}") print(f"{'Transformation':<25} {'Process data':<40}") print(f"{'Sink':<25} {'Data output':<40}") print(f"{'DataStream':<25} {'Unbounded stream':<40}") print(f"{'KeyedStream':<25} {'Partitioned by key':<40}") print(f"{'Parallelism':<25} {'Parallel instances':<40}")

💾 State Management

🎯 Managing State at Scale

State is data that operators remember across events. Flink provides managed state that is fault-tolerant, scalable, and queryable.

State Management

Keyed State: Per key. Operator State: Per operator. State Backend: Storage. Checkpoint: Snapshot. Queryable: Query state. TTL: Expiration.

💾 Keyed State💾 Operator State💾 State Backend💾 Checkpoint
# State Management in Flink print("=== What is State? ===") print(""" State: - Data remembered across events - Essential for stream processing - Must be fault-tolerant - Must be scalable Examples: - Running count - Last seen value - Session data - Machine learning model """) print("\n=== Types of State ===") print(""" 1. KEYED STATE - Per key - After keyBy - Partitioned by key - Types: - ValueState - ListState - MapState - ReducingState - AggregatingState 2. OPERATOR STATE - Per operator - Not keyed - Types: - ListState - BroadcastState - UnionListState 3. BROADCAST STATE - Broadcast to all tasks - Read-only - For rules/configs """) print("\n=== State Backends ===") print(""" State Backend: - Where state is stored - How it's checkpointed Options: 1. HashMapStateBackend - In-memory - Fast - Limited by RAM 2. EmbeddedRocksDBStateBackend - RocksDB (disk) - Larger state - Slower Configuration: env.set_state_backend(HashMapStateBackend()) env.set_state_backend(EmbeddedRocksDBStateBackend()) """) print("\n=== State TTL ===") print(""" TTL (Time-to-Live): - Expire state after time - Prevent unbounded growth - Configurable Example: StateTtlConfig ttlConfig = StateTtlConfig .newBuilder(Time.days(1)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build(); valueStateDescriptor.enableTimeToLive(ttlConfig); """) # IMPLEMENTATION print("\n=== Simulating State ===") class ValueState: """Simulate Flink ValueState.""" def __init__(self, default=None): self.value = default self.default = default def value(self): return self.value def update(self, value): self.value = value def clear(self): self.value = self.default class ListState: """Simulate Flink ListState.""" def __init__(self): self.values = [] def add(self, value): self.values.append(value) def get(self): return self.values def update(self, values): self.values = list(values) def clear(self): self.values = [] class MapState: """Simulate Flink MapState.""" def __init__(self): self.map = {} def put(self, key, value): self.map[key] = value def get(self, key): return self.map.get(key) def remove(self, key): if key in self.map: del self.map[key] def contains(self, key): return key in self.map def items(self): return self.map.items() def clear(self): self.map = {} class KeyedStateStore: """Simulate keyed state store.""" def __init__(self): self.states = defaultdict(dict) # state_name -> {key -> state} def get_state(self, state_name, key, state_type=ValueState): """Get or create state for key.""" if key not in self.states[state_name]: self.states[state_name][key] = state_type() return self.states[state_name][key] def clear_key(self, key): """Clear all state for key.""" for state_name in self.states: if key in self.states[state_name]: del self.states[state_name][key] def get_stats(self): """Get state statistics.""" stats = {} for state_name, keys in self.states.items(): stats[state_name] = len(keys) return stats # Demo print("=== State Demo ===") # Create state store store = KeyedStateStore() # Simulate events events = [ ('user1', 'click'), ('user2', 'view'), ('user1', 'click'), ('user3', 'purchase'), ('user2', 'click'), ('user1', 'view'), ('user2', 'purchase'), ] print("Processing events with state:") for user_id, action in events: # Get count state for user count_state = store.get_state('action_count', user_id, ValueState) count_state.update(count_state.value() or 0) # Get action list state action_list = store.get_state('actions', user_id, ListState) action_list.add(action) print(f" {user_id}: {action} (count={count_state.value()}, " f"actions={action_list.get()})") print("\nFinal state:") for state_name, keys in store.states.items(): print(f" {state_name}:") for key, state in keys.items(): if isinstance(state, ValueState): print(f" {key}: {state.value()}") elif isinstance(state, ListState): print(f" {key}: {state.get()}") print(f"\nState stats: {store.get_stats()}") # STATE SUMMARY print("\n=== State Summary ===") print(f"{'State Type':<25} {'Description':<40}") print("-" * 65) print(f"{'ValueState':<25} {'Single value per key':<40}") print(f"{'ListState':<25} {'List per key':<40}") print(f"{'MapState':<25} {'Map per key':<40}") print(f"{'ReducingState':<25} {'Aggregated value':<40}") print(f"{'BroadcastState':<25} {'Broadcast to all':<40}") print(f"{'OperatorState':<25} {'Per operator':<40}")

⏰ Time & Watermarks

🎯 Handling Time in Streams

Time is crucial in stream processing. Flink supports event-time, processing-time, and ingestion-time. Watermarks track event-time progress.

Time Semantics

Event-time: When event occurred. Processing-time: When processed. Ingestion-time: When ingested. Watermark: Event-time progress. Late data: After watermark. Allowed lateness: Grace period.

⏰ Event-time⏰ Processing-time⏰ Watermark⏰ Late Data
# Time and Watermarks in Flink print("=== Time Semantics ===") print(""" 1. EVENT-TIME - Time when event occurred - Embedded in event - Handle out-of-order - Deterministic 2. PROCESSING-TIME - Time when processed - System clock - Simple - Non-deterministic 3. INGESTION-TIME - Time when ingested - Source timestamp - Between event and processing """) print("\n=== Watermarks ===") print(""" Watermark: - Event-time progress indicator - "No more events before time T" - Triggers windows - Handles late data Example: Events: (event1, time=10) (event2, time=15) (event3, time=12) (watermark, time=20) Watermark 20 means: "No more events before time 20" (but event3 at 12 is late) """) print("\n=== Watermark Strategies ===") print(""" 1. PERIODIC WATERMARKS - Emitted periodically - Based on max event time - Default env.get_config().set_auto_watermark_interval(1000) 2. PUNCTUATED WATERMARKS - Emitted based on events - Custom logic 3. BOUNDED OUT-OF-ORDERNESS - Max delay - Watermark = max_time - delay WatermarkStrategy .for_bounded_out_of_orderness(Duration.seconds(5)) """) print("\n=== Allowed Lateness ===") print(""" Allowed Lateness: - Grace period for late data - After watermark - Update window results Example: window.allowed_lateness(Time.minutes(5)) Events: window: [10:00, 10:05) watermark: 10:06 late event: 10:04 → accepted (within 5 min) late event: 10:00 → rejected (too late) """) # IMPLEMENTATION print("\n=== Simulating Time & Watermarks ===") class Event: """Stream event with timestamp.""" def __init__(self, key, value, event_time): self.key = key self.value = value self.event_time = event_time self.processing_time = time.time() class Watermark: """Watermark for event-time progress.""" def __init__(self, timestamp): self.timestamp = timestamp def __repr__(self): return f"Watermark({self.timestamp})" class WatermarkGenerator: """Generate watermarks.""" def __init__(self, max_out_of_orderness=5): self.max_out_of_orderness = max_out_of_orderness self.max_event_time = 0 def process_event(self, event): """Process event, update max time.""" self.max_event_time = max(self.max_event_time, event.event_time) def generate_watermark(self): """Generate watermark.""" if self.max_event_time == 0: return None return Watermark(self.max_event_time - self.max_out_of_orderness) # Demo print("=== Watermark Demo ===") generator = WatermarkGenerator(max_out_of_orderness=5) events = [ Event('user1', 'click', 10), Event('user2', 'view', 15), Event('user1', 'click', 12), # Out of order Event('user3', 'purchase', 20), Event('user2', 'click', 18), # Out of order Event('user1', 'view', 25), ] print("Processing events:") for event in events: generator.process_event(event) wm = generator.generate_watermark() print(f" Event: key={event.key}, time={event.event_time}, " f"watermark={wm.timestamp if wm else 'none'}") print(f"\nFinal watermark: {generator.generate_watermark()}") # TIME SUMMARY print("\n=== Time Summary ===") print(f"{'Time Type':<25} {'Description':<40}") print("-" * 65) print(f"{'Event-time':<25} {'When event occurred':<40}") print(f"{'Processing-time':<25} {'When processed':<40}") print(f"{'Ingestion-time':<25} {'When ingested':<40}") print(f"{'Watermark':<25} {'Event-time progress':<40}") print(f"{'Allowed lateness':<25} {'Grace for late data':<40}")

🪟 Windows

🎯 Grouping Events by Time

Windows group events into finite buckets for aggregation. Flink supports tumbling, sliding, session, and custom windows.

Window Types

Tumbling: Fixed, non-overlapping. Sliding: Fixed, overlapping. Session: Activity-based. Global: All events. Count: By count. Custom: User-defined.

🪟 Tumbling🪟 Sliding🪟 Session🪟 Global
# Windows in Flink print("=== Window Types ===") print(""" 1. TUMBLING WINDOW - Fixed size - Non-overlapping - Each event in one window stream.window(TumblingEventTimeWindows.of(Time.seconds(5))) Windows: [0-5), [5-10), [10-15), ... 2. SLIDING WINDOW - Fixed size - Overlapping - Events in multiple windows stream.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))) Windows: [0-10), [5-15), [10-20), ... 3. SESSION WINDOW - Activity-based - Gap between events - No fixed size stream.window(EventTimeSessionWindows.withGap(Time.minutes(5))) Session: events within 5 min of each other 4. GLOBAL WINDOW - All events - Custom trigger stream.window(GlobalWindows.create()) """) print("\n=== Window Operations ===") print(""" Operations: - reduce: Aggregate - aggregate: Custom aggregation - apply: Process window - sum, min, max, count Example: stream .keyBy(event -> event.user) .window(TumblingEventTimeWindows.of(Time.seconds(5))) .sum("amount") """) print("\n=== Triggers ===") print(""" Trigger: - When to evaluate window - Default: watermark passes window end Types: - EventTimeTrigger - ProcessingTimeTrigger - CountTrigger - Custom Example: .trigger(CountTrigger.of(100)) (Evaluate every 100 events) """) print("\n=== Evictors ===") print(""" Evictor: - Remove elements from window - Before/after processing Types: - CountEvictor - TimeEvictor - DeltaEvictor Example: .evictor(CountEvictor.of(100)) (Keep only 100 elements) """) # IMPLEMENTATION print("\n=== Simulating Windows ===") class Window: """Base window.""" def __init__(self, start, end): self.start = start self.end = end self.elements = [] def add(self, element): self.elements.append(element) def __repr__(self): return f"Window[{self.start}, {self.end})" class TumblingWindow: """Tumbling window assigner.""" def __init__(self, size): self.size = size def assign(self, event): """Assign event to window.""" start = (event.event_time // self.size) * self.size return Window(start, start + self.size) class SlidingWindow: """Sliding window assigner.""" def __init__(self, size, slide): self.size = size self.slide = slide def assign(self, event): """Assign event to windows.""" windows = [] # Find all windows containing this event start = (event.event_time // self.slide) * self.slide for i in range(self.size // self.slide): w_start = start - i * self.slide w_end = w_start + self.size if w_start <= event.event_time < w_end: windows.append(Window(w_start, w_end)) return windows class SessionWindow: """Session window assigner.""" def __init__(self, gap): self.gap = gap def assign(self, events): """Assign events to sessions.""" if not events: return [] sorted_events = sorted(events, key=lambda e: e.event_time) sessions = [] current = Window(sorted_events[0].event_time, sorted_events[0].event_time + self.gap) for event in sorted_events[1:]: if event.event_time - current.end <= self.gap: current.end = event.event_time + self.gap current.add(event) else: sessions.append(current) current = Window(event.event_time, event.event_time + self.gap) sessions.append(current) return sessions # Demo print("=== Window Demo ===") # Create events events = [ Event('user1', 10, 10), Event('user2', 20, 12), Event('user1', 15, 15), Event('user2', 25, 18), Event('user1', 30, 22), Event('user2', 35, 25), Event('user1', 40, 28), ] # Tumbling windows print("1. Tumbling windows (size=10):") tumbling = TumblingWindow(10) windows = defaultdict(list) for event in events: w = tumbling.assign(event) windows[(w.start, w.end)].append(event) for (start, end), evts in sorted(windows.items()): print(f" Window [{start}, {end}):") for e in evts: print(f" {e.key}: {e.value}") # Sliding windows print("\n2. Sliding windows (size=10, slide=5):") sliding = SlidingWindow(10, 5) windows = defaultdict(list) for event in events: for w in sliding.assign(event): windows[(w.start, w.end)].append(event) for (start, end), evts in sorted(windows.items()): print(f" Window [{start}, {end}): {len(evts)} events") # Session windows print("\n3. Session windows (gap=3):") session = SessionWindow(3) sessions = session.assign(events) for s in sessions: print(f" Session [{s.start}, {s.end}): {len(s.elements)} events") # WINDOW SUMMARY print("\n=== Window Summary ===") print(f"{'Window':<25} {'Size':<20} {'Overlap':<15} {'Trigger':<15}") print("-" * 75) print(f"{'Tumbling':<25} {'Fixed':<20} {'No':<15} {'Watermark':<15}") print(f"{'Sliding':<25} {'Fixed':<20} {'Yes':<15} {'Watermark':<15}") print(f"{'Session':<25} {'Variable':<20} {'No':<15} {'Gap':<15}") print(f"{'Global':<25} {'All':<20} {'N/A':<15} {'Custom':<15}")

📸 Checkpoints & Recovery

🎯 Fault Tolerance

Checkpoints are periodic snapshots of state that enable recovery from failures. Flink uses the Chandy-Lamport algorithm for distributed snapshots.

Checkpoints

Checkpoint: Automatic snapshot. Savepoint: Manual checkpoint. Barrier: Snapshot marker. State: Operator state. Recovery: Restore from checkpoint. Exactly-once: Guaranteed.

📸 Checkpoint📸 Savepoint📸 Barrier📸 Recovery
# Checkpoints and Recovery print("=== What is a Checkpoint? ===") print(""" Checkpoint: - Periodic snapshot - Includes state - Includes offsets - Enables recovery - Automatic Savepoint: - Manual checkpoint - For upgrades - For migration - User-triggered """) print("\n=== Checkpoint Algorithm ===") print(""" Chandy-Lamport: 1. Source injects barrier 2. Barrier flows through DAG 3. Operators snapshot state 4. Barrier reaches sinks 5. Checkpoint complete Barrier: - Marker in stream - Aligns with events - Ensures consistency """) print("\n=== Checkpoint Configuration ===") print(""" # Enable checkpoints env.enable_checkpointing(10000) # Every 10 seconds # Checkpoint mode env.get_checkpoint_config().set_checkpointing_mode( CheckpointingMode.EXACTLY_ONCE ) # Timeout env.get_checkpoint_config().set_checkpoint_timeout(60000) # Min pause between checkpoints env.get_checkpoint_config().set_min_pause_between_checkpoints(5000) # Max concurrent checkpoints env.get_checkpoint_config().set_max_concurrent_checkpoints(1) # Externalized checkpoints env.get_checkpoint_config().enable_externalized_checkpoints( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION ) """) print("\n=== State Backends for Checkpoints ===") print(""" Checkpoints stored in: - JobManager memory - FileSystem (HDFS, S3) - RocksDB Configuration: env.get_checkpoint_config().set_checkpoint_storage( FileSystemCheckpointStorage("hdfs://checkpoints") ) """) print("\n=== Recovery ===") print(""" Recovery: 1. Job fails 2. JobManager detects 3. Restart from last checkpoint 4. Restore state 5. Resume from offsets Strategies: - Fixed delay - Exponential delay - Failure rate Configuration: env.set_restart_strategy( RestartStrategies.fixedDelayRestart(3, Time.seconds(10)) ) """) # IMPLEMENTATION print("\n=== Simulating Checkpoints ===") class Checkpoint: """Simulate a checkpoint.""" def __init__(self, checkpoint_id, timestamp): self.id = checkpoint_id self.timestamp = timestamp self.state = {} self.offsets = {} def snapshot(self, operator_id, state): """Snapshot operator state.""" self.state[operator_id] = state.copy() if hasattr(state, 'copy') else state def snapshot_offset(self, source_id, offset): """Snapshot source offset.""" self.offsets[source_id] = offset def __repr__(self): return f"Checkpoint(id={self.id}, state_keys={list(self.state.keys())}, offsets={self.offsets})" class CheckpointCoordinator: """Simulate checkpoint coordination.""" def __init__(self, interval_ms=10000): self.interval_ms = interval_ms self.checkpoints = [] self.current_id = 0 self.last_checkpoint_time = time.time() def should_checkpoint(self): """Check if checkpoint should be triggered.""" elapsed = (time.time() - self.last_checkpoint_time) * 1000 return elapsed >= self.interval_ms def trigger_checkpoint(self, operators, sources): """Trigger a checkpoint.""" self.current_id += 1 checkpoint = Checkpoint(self.current_id, time.time()) # Snapshot operators for op_id, state in operators.items(): checkpoint.snapshot(op_id, state) # Snapshot sources for src_id, offset in sources.items(): checkpoint.snapshot_offset(src_id, offset) self.checkpoints.append(checkpoint) self.last_checkpoint_time = time.time() return checkpoint def get_latest(self): """Get latest checkpoint.""" return self.checkpoints[-1] if self.checkpoints else None def get_stats(self): return { 'total': len(self.checkpoints), 'latest_id': self.current_id, 'interval_ms': self.interval_ms, } # Demo print("=== Checkpoint Demo ===") coordinator = CheckpointCoordinator(interval_ms=1000) # Simulate state operators = { 'count_op': {'user1': 5, 'user2': 3}, 'sum_op': {'user1': 100, 'user2': 200}, } sources = { 'kafka_source': 1000, # offset } # Trigger checkpoints for i in range(3): # Update state operators['count_op']['user1'] += 1 sources['kafka_source'] += 100 # Trigger checkpoint checkpoint = coordinator.trigger_checkpoint(operators, sources) print(f"\nCheckpoint {checkpoint.id}:") print(f" State: {checkpoint.state}") print(f" Offsets: {checkpoint.offsets}") # Show stats print(f"\nCoordinator stats: {coordinator.get_stats()}") # Simulate recovery print("\n=== Simulating Recovery ===") latest = coordinator.get_latest() print(f"Restoring from checkpoint {latest.id}") print(f" Restored state: {latest.state}") print(f" Restored offsets: {latest.offsets}") # CHECKPOINT SUMMARY print("\n=== Checkpoint Summary ===") print(f"{'Concept':<25} {'Description':<40}") print("-" * 65) print(f"{'Checkpoint':<25} {'Automatic snapshot':<40}") print(f"{'Savepoint':<25} {'Manual snapshot':<40}") print(f"{'Barrier':<25} {'Snapshot marker':<40}") print(f"{'State':<25} {'Operator state':<40}") print(f"{'Recovery':<25} {'Restore from checkpoint':<40}") print(f"{'Exactly-once':<25} {'Guaranteed':<40}")

🔌 Sources & Sinks

🎯 Connecting to External Systems

Sources read data into Flink; sinks write data out. Flink provides connectors for Kafka, files, databases, and more.

Sources & Sinks

Source: Data input. Sink: Data output. Kafka: Most common. File: HDFS, S3. Database: JDBC. Custom: User-defined.

🔌 Source🔌 Sink🔌 Kafka🔌 JDBC
# Sources and Sinks in Flink print("=== Common Sources ===") print(""" 1. KAFKA SOURCE - Most common - Exactly-once - Parallel KafkaSource source = KafkaSource.builder() .setBootstrapServers("localhost:9092") .setTopics("input") .setGroupId("flink-group") .setStartingOffsets(OffsetsInitializer.earliest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); 2. FILE SOURCE - HDFS, S3, local - Batch or streaming stream = env.read_text_file("hdfs://data/input.txt") 3. SOCKET SOURCE - Testing stream = env.socket_text_stream("localhost", 9999) 4. COLLECTION SOURCE - Testing stream = env.from_collection([1, 2, 3, 4, 5]) 5. JDBC SOURCE - Database stream = env.create_input(JDBCInputFormat.buildJDBCInputFormat()...) """) print("\n=== Common Sinks ===") print(""" 1. KAFKA SINK - Most common - Exactly-once KafkaSink sink = KafkaSink.builder() .setBootstrapServers("localhost:9092") .setRecordSerializer(KafkaRecordSerializationSchema.builder() .setTopic("output") .setValueSerializationSchema(new SimpleStringSchema()) .build()) .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) .build(); 2. FILE SINK - HDFS, S3 stream.write_as_text("hdfs://data/output") 3. JDBC SINK - Database stream.add_sink(JdbcSink.sink(...)) 4. ELASTICSEARCH SINK - Search stream.add_sink(ElasticsearchSink(...)) 5. CUSTOM SINK - User-defined """) print("\n=== Connectors ===") print(""" Official connectors: - Kafka - FileSystem - JDBC - Elasticsearch - HBase - Cassandra - RabbitMQ - AWS Kinesis - Google Pub/Sub Community connectors: - Redis - MongoDB - Neo4j - ... """) # IMPLEMENTATION print("\n=== Simulating Sources & Sinks ===") class Source: """Base source.""" def read(self): raise NotImplementedError class CollectionSource(Source): """Collection source.""" def __init__(self, data): self.data = data def read(self): return self.data class FileSource(Source): """File source.""" def __init__(self, path): self.path = path self.data = [] def read(self): # Simulate reading return self.data class KafkaSource(Source): """Kafka source.""" def __init__(self, topic, bootstrap_servers='localhost:9092'): self.topic = topic self.bootstrap_servers = bootstrap_servers self.messages = [] def read(self): return self.messages class Sink: """Base sink.""" def write(self, data): raise NotImplementedError class CollectionSink(Sink): """Collection sink.""" def __init__(self): self.data = [] def write(self, data): self.data.extend(data) return len(data) class KafkaSink(Sink): """Kafka sink.""" def __init__(self, topic, bootstrap_servers='localhost:9092'): self.topic = topic self.bootstrap_servers = bootstrap_servers self.messages = [] def write(self, data): self.messages.extend(data) return len(data) class FileSink(Sink): """File sink.""" def __init__(self, path): self.path = path self.data = [] def write(self, data): self.data.extend(data) return len(data) # Demo print("=== Sources & Sinks Demo ===") # Create source source = CollectionSource([1, 2, 3, 4, 5]) data = source.read() print(f"Source data: {data}") # Process processed = [x * 2 for x in data] print(f"Processed: {processed}") # Create sink sink = CollectionSink() sink.write(processed) print(f"Sink data: {sink.data}") # Kafka example print("\n=== Kafka Source & Sink ===") kafka_source = KafkaSource('input-topic') kafka_source.messages = [ '{"user": "Alice", "action": "click"}', '{"user": "Bob", "action": "view"}', ] messages = kafka_source.read() print(f"Kafka source messages: {len(messages)}") kafka_sink = KafkaSink('output-topic') kafka_sink.write(messages) print(f"Kafka sink messages: {len(kafka_sink.messages)}") # SOURCE/SINK SUMMARY print("\n=== Source & Sink Summary ===") print(f"{'Type':<25} {'Source':<25} {'Sink':<25}") print("-" * 75) print(f"{'Kafka':<25} {'Yes':<25} {'Yes':<25}") print(f"{'File':<25} {'Yes':<25} {'Yes':<25}") print(f"{'JDBC':<25} {'Yes':<25} {'Yes':<25}") print(f"{'Socket':<25} {'Yes':<25} {'No':<25}") print(f"{'Collection':<25} {'Yes':<25} {'No':<25}") print(f"{'Elasticsearch':<25} {'No':<25} {'Yes':<25}")

🗄️ Table API & SQL

🎯 SQL on Streams

Flink's Table API and SQL provide a relational interface for stream and batch processing. Write SQL queries over streaming data.

Table API & SQL

Table: Relational abstraction. SQL: Standard queries. Dynamic: Tables change over time. Unified: Batch + stream. Connectors: Kafka, files. UDFs: Custom functions.

🗄️ Table API🗄️ SQL🗄️ Dynamic Tables🗄️ Unified
# Table API & SQL print("=== Table API Overview ===") print(""" Table API: - Relational abstraction - Similar to DataFrames - Declarative - Optimized SQL: - Standard SQL - Continuous queries - Dynamic tables - Unified batch + stream Example: TableEnvironment tEnv = TableEnvironment.create(settings); // Create table from Kafka tEnv.executeSql( "CREATE TABLE orders (" + " order_id INT," + " user_id INT," + " amount DOUBLE," + " order_time TIMESTAMP(3)," + " WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND" + ") WITH (" + " 'connector' = 'kafka'," + " 'topic' = 'orders'," + " 'properties.bootstrap.servers' = 'localhost:9092'," + " 'format' = 'json'" + ")" ); // Query Table result = tEnv.sqlQuery( "SELECT user_id, SUM(amount) as total " + "FROM orders " + "GROUP BY user_id" ); // Output result.executeInsert("output_table"); """) print("\n=== Dynamic Tables ===") print(""" Dynamic Table: - Changes over time - Like a materialized view - Updated by stream Operations: - SELECT - JOIN - GROUP BY - WINDOW Example: -- Tumbling window SELECT TUMBLE_START(order_time, INTERVAL '1' HOUR) as window_start, user_id, COUNT(*) as order_count, SUM(amount) as total FROM orders GROUP BY TUMBLE(order_time, INTERVAL '1' HOUR), user_id """) print("\n=== SQL Operations ===") print(""" SELECT: SELECT * FROM orders WHERE amount > 100 JOIN: SELECT o.order_id, u.name FROM orders o JOIN users u ON o.user_id = u.id GROUP BY: SELECT user_id, COUNT(*) FROM orders GROUP BY user_id WINDOW: SELECT TUMBLE_START(order_time, INTERVAL '1' HOUR), COUNT(*) FROM orders GROUP BY TUMBLE(order_time, INTERVAL '1' HOUR) OVER: SELECT order_id, amount, SUM(amount) OVER (PARTITION BY user_id ORDER BY order_time) FROM orders """) # IMPLEMENTATION print("\n=== Simulating Table API ===") class DynamicTable: """Simulate Flink Dynamic Table.""" def __init__(self, name, data=None): self.name = name self.data = data or [] self.changelog = [] def insert(self, row): """Insert row (upsert).""" self.data.append(row) self.changelog.append(('+I', row)) def update(self, key, updates): """Update row.""" for row in self.data: if row.get('id') == key: old = row.copy() row.update(updates) self.changelog.append(('-U', old)) self.changelog.append(('+U', row)) return True return False def delete(self, key): """Delete row.""" for i, row in enumerate(self.data): if row.get('id') == key: self.changelog.append(('-D', row)) del self.data[i] return True return False def select(self, columns='*', where=None, group_by=None, aggregates=None): """Execute SELECT.""" data = self.data # WHERE if where: col, op, val = where if op == '>': data = [r for r in data if r.get(col, 0) > val] elif op == '<': data = [r for r in data if r.get(col, 0) < val] elif op == '=': data = [r for r in data if r.get(col) == val] # GROUP BY if group_by: grouped = defaultdict(list) for row in data: grouped[row[group_by]].append(row) result = [] for key, rows in grouped.items(): row_result = {group_by: key} if aggregates: for name, func, col in aggregates: values = [r[col] for r in rows] if func == 'COUNT': row_result[name] = len(values) elif func == 'SUM': row_result[name] = sum(values) elif func == 'AVG': row_result[name] = sum(values) / len(values) result.append(row_result) data = result # SELECT columns if columns != '*': col_list = [c.strip() for c in columns.split(',')] data = [{k: r.get(k) for k in col_list} for r in data] return data def show(self): """Show table.""" print(f"Table: {self.name}") for row in self.data: print(f" {row}") def get_changelog(self): """Get changelog.""" return self.changelog # Demo print("=== Table API Demo ===") # Create table orders = DynamicTable('orders') # Insert data orders.insert({'id': 1, 'user_id': 101, 'amount': 100, 'status': 'created'}) orders.insert({'id': 2, 'user_id': 102, 'amount': 200, 'status': 'created'}) orders.insert({'id': 3, 'user_id': 101, 'amount': 150, 'status': 'created'}) orders.insert({'id': 4, 'user_id': 103, 'amount': 300, 'status': 'created'}) print("1. Initial table:") orders.show() print("\n2. SELECT with WHERE:") result = orders.select(where=('amount', '>', 100)) for row in result: print(f" {row}") print("\n3. GROUP BY user_id:") result = orders.select( group_by='user_id', aggregates=[ ('count', 'COUNT', 'id'), ('total', 'SUM', 'amount'), ] ) for row in result: print(f" {row}") print("\n4. Update (upsert):") orders.update(1, {'amount': 120}) orders.show() print("\n5. Changelog:") for op, row in orders.get_changelog(): print(f" {op}: {row}") # TABLE API SUMMARY print("\n=== Table API Summary ===") print(f"{'Feature':<25} {'Description':<40}") print("-" * 65) print(f"{'Table':<25} {'Relational abstraction':<40}") print(f"{'SQL':<25} {'Standard queries':<40}") print(f"{'Dynamic Table':<25} {'Changes over time':<40}") print(f"{'Unified':<25} {'Batch + stream':<40}") print(f"{'Connectors':<25} {'Kafka, files, JDBC':<40}") print(f"{'UDFs':<25} {'Custom functions':<40}")

🔍 Complex Event Processing

🎯 Pattern Detection in Streams

CEP (Complex Event Processing) detects patterns in event streams. FlinkCEP provides a library for pattern matching over streams.

CEP

Pattern: Sequence of events. Match: Find occurrences. Window: Within time. Quantifier: One or more. Condition: Filter. Action: Trigger.

🔍 Pattern🔍 Match🔍 Quantifier🔍 Window
# Complex Event Processing (CEP) print("=== What is CEP? ===") print(""" CEP: - Detect patterns in event streams - Sequence of events - Temporal relationships - Complex conditions Use cases: - Fraud detection - Intrusion detection - Monitoring - Alerting - Business rules Example: Detect: Login failure followed by login success within 5 minutes from different IPs → Potential security breach """) print("\n=== Pattern API ===") print(""" Pattern: - begin: Start pattern - next: Followed by - followedBy: Followed by (relaxed) - where: Condition - within: Time window - times: Quantifier Example: Pattern pattern = Pattern.begin("first") .where(new SimpleCondition() { public boolean filter(Event event) { return event.getType().equals("login_failure"); } }) .next("second") .where(new SimpleCondition() { public boolean filter(Event event) { return event.getType().equals("login_success"); } }) .within(Time.minutes(5)); PatternStream patternStream = CEP.pattern(input, pattern); """) print("\n=== Quantifiers ===") print(""" Quantifiers: - times(n): Exactly n times - times(n, m): Between n and m times - oneOrMore: 1 or more - zeroOrMore: 0 or more - optional: 0 or 1 Example: Pattern.begin("first") .where(...) .times(3) .next("second") .where(...) """) print("\n=== Pattern Types ===") print(""" 1. STRICT - next: Immediate - No events between 2. RELAXED - followedBy: Any events between - Loose 3. NOT - notNext, notFollowedBy - Negative patterns """) # IMPLEMENTATION print("\n=== Simulating CEP ===") class CEPEvent: """CEP event.""" def __init__(self, event_type, timestamp, data=None): self.type = event_type self.timestamp = timestamp self.data = data or {} def __repr__(self): return f"Event({self.type}, t={self.timestamp})" class CEPPattern: """Simulate CEP pattern.""" def __init__(self): self.steps = [] self.time_window = None def begin(self, name): self.steps.append({'name': name, 'type': 'begin', 'condition': None}) return self def next(self, name): self.steps.append({'name': name, 'type': 'next', 'condition': None}) return self def where(self, condition): if self.steps: self.steps[-1]['condition'] = condition return self def within(self, time_window): self.time_window = time_window return self def match(self, events): """Find pattern matches.""" matches = [] for i, event in enumerate(events): match = [event] step_idx = 1 for j in range(i + 1, len(events)): if step_idx >= len(self.steps): break next_event = events[j] step = self.steps[step_idx] # Check time window if self.time_window: if next_event.timestamp - event.timestamp > self.time_window: break # Check condition if step['condition'] and not step['condition'](next_event): continue match.append(next_event) step_idx += 1 if step_idx == len(self.steps): matches.append(match) return matches # Demo print("=== CEP Demo ===") # Create events events = [ CEPEvent('login_failure', 100, {'ip': '192.168.1.1'}), CEPEvent('login_success', 105, {'ip': '192.168.1.2'}), CEPEvent('login_failure', 200, {'ip': '192.168.1.3'}), CEPEvent('login_failure', 205, {'ip': '192.168.1.3'}), CEPEvent('login_success', 210, {'ip': '192.168.1.3'}), CEPEvent('purchase', 300, {'amount': 100}), CEPEvent('login_failure', 400, {'ip': '10.0.0.1'}), CEPEvent('login_success', 450, {'ip': '10.0.0.2'}), ] print("Events:") for e in events: print(f" {e}") # Define pattern: login_failure followed by login_success within 60s pattern = CEPPattern() pattern.begin('failure') \ .where(lambda e: e.type == 'login_failure') \ .next('success') \ .where(lambda e: e.type == 'login_success') \ .within(60) print("\nPattern: login_failure → login_success within 60s") matches = pattern.match(events) print(f"\nMatches found: {len(matches)}") for i, match in enumerate(matches): print(f" Match {i+1}:") for event in match: print(f" {event}") # CEP SUMMARY print("\n=== CEP Summary ===") print(f"{'Concept':<25} {'Description':<40}") print("-" * 65) print(f"{'Pattern':<25} {'Sequence of events':<40}") print(f"{'Match':<25} {'Occurrence of pattern':<40}") print(f"{'Quantifier':<25} {'Count of events':<40}") print(f"{'Window':<25} {'Time constraint':<40}") print(f"{'Condition':<25} {'Filter events':<40}") print(f"{'Action':<25} {'Trigger on match':<40}")

⚖️ Flink vs Spark

🎯 Comparing Stream Processors

Flink and Spark are both distributed processing frameworks. Flink is streaming-first with true event-at-a-time processing; Spark uses micro-batch for streaming.

Flink vs Spark

Flink: True streaming, low latency. Spark: Micro-batch, mature. Latency: Flink ms, Spark seconds. State: Flink better. Ecosystem: Spark broader. Use: Both.

⚖️ Streaming⚖️ Latency⚖️ State⚖️ Ecosystem
# Flink vs Spark print("=== Flink vs Spark ===") print(f"{'Aspect':<25} {'Flink':<25} {'Spark':<25}") print("-" * 75) print(f"{'Processing':<25} {'True streaming':<25} {'Micro-batch':<25}") print(f"{'Latency':<25} {'Milliseconds':<25} {'Seconds':<25}") print(f"{'Throughput':<25} {'High':<25} {'Very high':<25}") print(f"{'State':<25} {'Built-in':<25} {'Built-in':<25}") print(f"{'Event-time':<25} {'Native':<25} {'Supported':<25}") print(f"{'Exactly-once':<25} {'Yes':<25} {'Yes':<25}") print(f"{'Batch':<25} {'Bounded stream':<25} {'Native':<25}") print(f"{'SQL':<25} {'Table API/SQL':<25} {'Spark SQL':<25}") print(f"{'ML':<25} {'FlinkML':<25} {'MLlib':<25}") print(f"{'Graph':<25} {'Gelly':<25} {'GraphX':<25}") print(f"{'Maturity':<25} {'Growing':<25} {'Mature':<25}") print(f"{'Ecosystem':<25} {'Smaller':<25} {'Larger':<25}") print("\n=== When to Use Which ===") print(""" USE FLINK WHEN: ✓ True streaming (low latency) ✓ Event-time processing ✓ Complex state ✓ CEP ✓ Real-time analytics ✓ Event-driven apps USE SPARK WHEN: ✓ Batch processing ✓ ML pipelines ✓ Existing Spark code ✓ Unified analytics ✓ Larger ecosystem ✓ Interactive queries BOTH: ✓ Kafka integration ✓ SQL ✓ Exactly-once ✓ Scalable ✓ Fault-tolerant """) print("\n=== Architecture Comparison ===") print(""" FLINK: - Event-at-a-time - Pipelined execution - Low latency - Backpressure - State backend SPARK: - Micro-batch - DAG execution - Higher latency - RDD lineage - In-memory """) # PERFORMANCE COMPARISON print("\n=== Performance Comparison ===") print(f"{'Scenario':<30} {'Flink':<20} {'Spark':<20}") print("-" * 70) print(f"{'Latency':<30} {'1-10 ms':<20} {'100 ms - 1 s':<20}") print(f"{'Throughput':<30} {'Millions/sec':<20} {'Millions/sec':<20}") print(f"{'State size':<30} {'Terabytes':<20} {'Terabytes':<20}") print(f"{'Recovery':<30} {'Seconds':<20} {'Seconds':<20}") print(f"{'Batch':<30} {'Good':<20} {'Excellent':<20}") print(f"{'Streaming':<30} {'Excellent':<20} {'Good':<20}") # COMPARISON SUMMARY print("\n=== Comparison Summary ===") print(f"{'Feature':<25} {'Flink':<25} {'Spark':<25}") print("-" * 75) print(f"{'First release':<25} {'2014':<25} {'2014':<25}") print(f"{'Language':<25} {'Java, Scala, Python':<25} {'Scala, Python, Java, R':<25}") print(f"{'Streaming':<25} {'Native':<25} {'Micro-batch':<25}") print(f"{'Latency':<25} {'ms':<25} {'seconds':<25}") print(f"{'Batch':<25} {'Unified':<25} {'Native':<25}") print(f"{'SQL':<25} {'Yes':<25} {'Yes':<25}") print(f"{'ML':<25} {'FlinkML':<25} {'MLlib':<25}") print(f"{'Ecosystem':<25} {'Growing':<25} {'Mature':<25}")

💼 Applications

🎯 Flink in the Real World

Flink powers real-time analytics, event-driven applications, and streaming ETL at companies like Alibaba, Uber, and Netflix.

Applications

Analytics: Real-time. Fraud: Detection. IoT: Sensor data. ETL: Streaming. Monitoring: Alerts. Gaming: Analytics.

💼 Analytics💼 Fraud💼 IoT💼 ETL
# Flink Applications print("=== 1. Real-Time Analytics ===") print(""" Use case: - Clickstream analysis - User behavior - Real-time dashboards Pipeline: Events → Kafka → Flink → Dashboard → Alerts → Database Example: -- Count clicks per second stream .keyBy(event -> event.page) .window(TumblingEventTimeWindows.of(Time.seconds(1))) .count() .addSink(dashboardSink); """) print("\n=== 2. Fraud Detection ===") print(""" Use case: - Credit card fraud - Account takeover - Money laundering Pipeline: Transactions → Kafka → Flink CEP → Alerts → Block → Report Example: -- Detect suspicious pattern Pattern.begin("small") .where(t -> t.amount < 1) .next("large") .where(t -> t.amount > 10000) .within(Time.minutes(5)); """) print("\n=== 3. IoT Analytics ===") print(""" Use case: - Sensor data - Predictive maintenance - Anomaly detection Pipeline: Sensors → Kafka → Flink → Alerts → Dashboard → Data Lake Example: -- Detect temperature spike stream .keyBy(sensor -> sensor.id) .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1))) .aggregate(new TemperatureAggregate()) .filter(avg -> avg > 100) .addSink(alertSink); """) print("\n=== 4. Streaming ETL ===") print(""" Use case: - Data pipelines - Data enrichment - Data cleaning Pipeline: Source → Flink (transform) → Sink Example: -- Clean and enrich stream .filter(event -> event.isValid()) .map(event -> enrich(event)) .keyBy(event -> event.userId) .addSink(dataWarehouseSink); """) print("\n=== 5. Monitoring and Alerting ===") print(""" Use case: - System monitoring - Application monitoring - Security monitoring Pipeline: Metrics → Kafka → Flink → Alerts → Dashboard → PagerDuty Example: -- Alert on error rate stream .filter(log -> log.level == "ERROR") .window(TumblingEventTimeWindows.of(Time.minutes(5))) .count() .filter(count -> count > 100) .addSink(alertSink); """) print("\n=== 6. Gaming Analytics ===") print(""" Use case: - Player behavior - Game economy - Matchmaking Pipeline: Game events → Kafka → Flink → Analytics → Leaderboard → Recommendations Example: -- Real-time leaderboard stream .keyBy(event -> event.playerId) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .sum("score") .addSink(leaderboardSink); """) # IMPLEMENTATION print("\n=== Practical Example: Fraud Detection ===") class Transaction: """Transaction event.""" def __init__(self, user_id, amount, timestamp, location): self.user_id = user_id self.amount = amount self.timestamp = timestamp self.location = location def __repr__(self): return f"Transaction(user={self.user_id}, amount={self.amount}, loc={self.location})" class FraudDetector: """Simple fraud detector using Flink-like patterns.""" def __init__(self, threshold=10000, time_window=300): self.threshold = threshold self.time_window = time_window self.alerts = [] def process(self, transactions): """Process transactions for fraud.""" for i, txn in enumerate(transactions): # Check for high amount if txn.amount > self.threshold: self.alerts.append({ 'type': 'HIGH_AMOUNT', 'transaction': txn, 'reason': f'Amount {txn.amount} > {self.threshold}', }) # Check for rapid transactions for j in range(max(0, i-5), i): prev = transactions[j] if (prev.user_id == txn.user_id and txn.timestamp - prev.timestamp < 60 and txn.location != prev.location): self.alerts.append({ 'type': 'LOCATION_CHANGE', 'transaction': txn, 'previous': prev, 'reason': 'Location change within 60s', }) return self.alerts # Demo print("=== Fraud Detection Demo ===") detector = FraudDetector(threshold=10000) transactions = [ Transaction('user1', 100, 1000, 'NYC'), Transaction('user1', 150, 1030, 'NYC'), Transaction('user2', 500, 1040, 'LA'), Transaction('user1', 20000, 1050, 'NYC'), # High amount Transaction('user2', 300, 1060, 'LA'), Transaction('user1', 5000, 1070, 'London'), # Location change Transaction('user3', 15000, 1080, 'Chicago'), # High amount ] print("Transactions:") for txn in transactions: print(f" {txn}") alerts = detector.process(transactions) print(f"\nAlerts: {len(alerts)}") for alert in alerts: print(f" [{alert['type']}] {alert['reason']}") print(f" Transaction: {alert['transaction']}") # APPLICATION SUMMARY print("\n=== Application Summary ===") print(f"{'Domain':<25} {'Use Case':<30} {'Latency':<15}") print("-" * 70) print(f"{'Analytics':<25} {'Real-time dashboards':<30} {'ms':<15}") print(f"{'Fraud':<25} {'Detection':<30} {'ms':<15}") print(f"{'IoT':<25} {'Anomaly detection':<30} {'ms':<15}") print(f"{'ETL':<25} {'Streaming pipelines':<30} {'ms':<15}") print(f"{'Monitoring':<25} {'Alerting':<30} {'ms':<15}") print(f"{'Gaming':<25} {'Leaderboards':<30} {'ms':<15}")

⚡ Performance & Tuning

🎯 Making Flink Fast

Flink performance depends on parallelism, state backend, checkpointing, and network tuning. Key: balance resources and minimize overhead.

Performance Tuning

Parallelism: More = faster. State backend: Memory vs disk. Checkpoint: Interval. Network: Buffers. Backpressure: Monitor. Resources: CPU, memory.

⚡ Parallelism⚡ State Backend⚡ Checkpoint⚡ Backpressure
# Flink Performance Tuning print("=== Performance Factors ===") print(""" 1. PARALLELISM - Number of parallel instances - More = more throughput - Limited by resources 2. STATE BACKEND - HashMap: Fast, RAM-limited - RocksDB: Larger, disk-based 3. CHECKPOINTING - Interval - Timeout - Storage 4. NETWORK - Buffers - Compression - Batching 5. RESOURCES - CPU - Memory - Disk """) print("\n=== Parallelism Tuning ===") print(""" Set parallelism: env.set_parallelism(8) Operator-level: stream.map(...).set_parallelism(4) Max parallelism: env.set_max_parallelism(128) Guidelines: - Match cores - 1-2 per core - Balance operators """) print("\n=== State Backend Tuning ===") print(""" HashMapStateBackend: - In-memory - Fast - Limited by RAM - Use for: small state EmbeddedRocksDBStateBackend: - Disk-based - Larger state - Slower - Use for: large state Configuration: env.set_state_backend(HashMapStateBackend()) env.set_state_backend(EmbeddedRocksDBStateBackend(true)) """) print("\n=== Checkpoint Tuning ===") print(""" Interval: - Balance overhead vs recovery time - 1-10 minutes typical - Shorter = more overhead Timeout: - Max time for checkpoint - 10 minutes default Min pause: - Between checkpoints - Avoid overlap Configuration: env.enable_checkpointing(60000) env.get_checkpoint_config().set_checkpoint_timeout(300000) env.get_checkpoint_config().set_min_pause_between_checkpoints(30000) """) print("\n=== Network Tuning ===") print(""" Buffers: taskmanager.network.memory.fraction: 0.1 taskmanager.network.memory.min: 64mb taskmanager.network.memory.max: 1gb Compression: -- Enable execution.checkpointing.compression: true Batching: - Automatic - Tune buffer timeout """) print("\n=== Backpressure ===") print(""" Backpressure: - Consumer slower than producer - Causes slowdown - Monitor in UI Causes: - Slow operator - Insufficient parallelism - Network bottleneck - Disk I/O Solutions: - Increase parallelism - Optimize operator - More resources - Better partitioning """) # IMPLEMENTATION print("\n=== Simulating Performance ===") class FlinkPerformanceSimulator: """Simulate Flink performance.""" def __init__(self, parallelism=4, state_backend='hashmap'): self.parallelism = parallelism self.state_backend = state_backend self.metrics = { 'throughput': 0, 'latency': 0, 'checkpoint_time': 0, } def simulate(self, events_per_sec, state_size_mb, checkpoint_interval_ms): """Simulate performance.""" print(f"\n=== Simulation ===") print(f"Parallelism: {self.parallelism}") print(f"State backend: {self.state_backend}") print(f"Events/sec: {events_per_sec}") print(f"State size: {state_size_mb} MB") print(f"Checkpoint interval: {checkpoint_interval_ms} ms") # Throughput (limited by parallelism) max_throughput = self.parallelism * 100000 # 100k per instance actual_throughput = min(events_per_sec, max_throughput) # Latency base_latency = 1 # 1ms if self.state_backend == 'rocksdb': base_latency += 2 # Disk access latency = base_latency # Checkpoint time if self.state_backend == 'hashmap': checkpoint_time = state_size_mb / 100 # 100 MB/s else: checkpoint_time = state_size_mb / 50 # 50 MB/s # Checkpoint overhead checkpoint_overhead = (checkpoint_time / checkpoint_interval_ms) * 100 self.metrics = { 'throughput': actual_throughput, 'latency': latency, 'checkpoint_time': checkpoint_time, 'checkpoint_overhead': checkpoint_overhead, } print(f"\nResults:") print(f" Throughput: {actual_throughput:,} events/sec") print(f" Latency: {latency:.1f} ms") print(f" Checkpoint time: {checkpoint_time:.1f} ms") print(f" Checkpoint overhead: {checkpoint_overhead:.1f}%") return self.metrics # Demo print("=== Performance Simulation ===") # Low parallelism sim1 = FlinkPerformanceSimulator(parallelism=2, state_backend='hashmap') sim1.simulate(events_per_sec=100000, state_size_mb=100, checkpoint_interval_ms=60000) # High parallelism sim2 = FlinkPerformanceSimulator(parallelism=16, state_backend='hashmap') sim2.simulate(events_per_sec=1000000, state_size_mb=100, checkpoint_interval_ms=60000) # RocksDB for large state sim3 = FlinkPerformanceSimulator(parallelism=16, state_backend='rocksdb') sim3.simulate(events_per_sec=1000000, state_size_mb=10000, checkpoint_interval_ms=60000) # PERFORMANCE SUMMARY print("\n=== Performance Summary ===") print(f"{'Factor':<25} {'Impact':<30} {'Recommendation':<25}") print("-" * 80) print(f"{'Parallelism':<25} {'High':<30} {'Match cores':<25}") print(f"{'State backend':<25} {'Medium':<30} {'HashMap for small':<25}") print(f"{'Checkpoint interval':<25} {'Medium':<30} {'1-10 minutes':<25}") print(f"{'Network buffers':<25} {'Low':<30} {'Tune if needed':<25}") print(f"{'Backpressure':<25} {'High':<30} {'Monitor and fix':<25}")

🚀 Future: Flink in Modern Computing

🎯 Where Flink Is Going

Flink is evolving: unified batch/stream, cloud-native, AI integration, and better SQL. It's becoming the standard for stream processing.

Future Directions

Unified: Batch + stream. Cloud: Managed services. SQL: Better SQL. AI: ML pipelines. Kubernetes: Native. State: Better state.

🚀 Unified🚀 Cloud🚀 SQL🚀 AI
# Future of Apache Flink print("=== 1. Unified Batch and Stream ===") print(""" Trend: One API for batch and stream Current: - DataStream API (stream) - DataSet API (batch, legacy) - Table API/SQL (unified) Future: - DataStream API for both - Bounded streams = batch - Same code, same runtime Benefits: - Simpler - Consistent - Unified semantics """) print("\n=== 2. Cloud-Native Flink ===") print(""" Trend: Flink on cloud Managed services: - AWS Kinesis Data Analytics - Confluent Cloud - Ververica Platform - Azure HDInsight Kubernetes: - Flink Operator - Native integration - Auto-scaling Serverless: - No infrastructure - Pay per use - Event-driven Future: - Cloud-native - Serverless - Kubernetes """) print("\n=== 3. SQL Improvements ===") print(""" Trend: SQL-first Current: - Table API - SQL - Connectors Future: - Better SQL - More connectors - Materialized views - Streaming SQL Use cases: - Analytics - ETL - Monitoring - Alerting """) print("\n=== 4. AI and ML ===") print(""" Trend: Flink for ML Current: - FlinkML - Integration with TensorFlow, PyTorch Future: - Real-time ML - Feature stores - Model serving - Online learning Use cases: - Fraud detection - Recommendations - Predictive maintenance """) print("\n=== 5. State Improvements ===") print(""" Trend: Better state management Current: - HashMap - RocksDB - Checkpoints Future: - Disaggregated state - Tiered storage - Better performance - Larger state Benefits: - Scalability - Cost - Performance """) print("\n=== 6. Edge and IoT ===") print(""" Trend: Flink at edge Current: - Cloud-based - Centralized Future: - Edge Flink - IoT processing - Real-time decisions - Offline capability Use cases: - Autonomous vehicles - Smart factories - Remote monitoring """) # TIMELINE print("\n=== Timeline ===") print(""" Past: 2014: Flink started (Stratosphere) 2015: Apache incubator 2016: Flink 1.0 2017: Flink 1.4 (SQL) 2019: Flink 1.9 (Table API) 2020: Flink 1.11 (unified) 2022: Flink 1.15 Present: 2024: Flink 1.19 2025: Unified batch/stream Future: 2030: Cloud-native? 2035: AI-native? 2040: Edge-native? """) print("\n=== Final Thoughts ===") print(""" Apache Flink is: - Streaming-first - Stateful - Event-time - Exactly-once - Evolving The future is: - Unified batch/stream - Cloud-native - SQL-first - AI-ready - Edge-enabled "Flink is the future of stream processing." Keep exploring. Keep building. """)