Skip to content

Phase 1: Distributed Architecture, Dataflow Compilation & Runtime Model

Audience: Principal Distributed Systems Architect & Streaming Engine Specialist
Source References: flink-runtime/src/main/java/org/apache/flink/runtime/, flink-streaming-java/

1. Process Hierarchy: JobManager vs. TaskManager

An Apache Flink cluster operates under a master-worker distributed architecture:

┌────────────────────────────────────────────────────────────────────────┐
│ JobManager (Cluster Master Process)                                    │
├────────────────────────────────────────────────────────────────────────┤
│ 1. Dispatcher: HTTP/REST endpoint; accepts client JAR submissions      │
│ 2. ResourceManager: Manages TaskManager registration & slot allocation │
│ 3. JobMaster: One per active job; compiles & drives the ExecutionGraph │
│ 4. CheckpointCoordinator: Injects barriers & validates global snapshots│
└───────────────────────────────────┬────────────────────────────────────┘
                                    │ RPC / Heartbeat
         ┌──────────────────────────┴──────────────────────────┐
         ▼                                                     ▼
┌────────────────────────────────────┐       ┌───────────────────────────────────┐
│ TaskManager 1 (Worker JVM Process) │       │ TaskManager 2 (Worker JVM Process)│
├────────────────────────────────────┤       ├───────────────────────────────────┤
│ Slot 1 (Thread)  │ Slot 2 (Thread) │       │ Slot 1 (Thread)  │ Slot 2 (Thread)│
├────────────────────────────────────┤       ├───────────────────────────────────┤
│ Managed Memory Pool (RocksDB)      │       │ Managed Memory Pool (RocksDB)     │
│ NetworkBufferPool (Netty)          │       │ NetworkBufferPool (Netty)         │
└────────────────────────────────────┘       └───────────────────────────────────┘

1.1 Task Slots & Resource Isolation

  • A Task Slot is a fixed slice of resources inside a TaskManager JVM.
  • If a TaskManager has 4 slots, it allocates 25% of the managed memory pool to each slot.
  • Crucial: Task slots share the same JVM heap and OS process. Slot isolation is strictly memory-based; CPU cores are not hard-isolated unless deployed via Kubernetes cgroups.

2. The 4-Stage Graph Compilation Pipeline

Before execution, user-defined transformations are transformed through four distinct graph representations:

StreamGraph ──► JobGraph ──► ExecutionGraph ──► Physical Execution Tasks
 (Client)        (Client)     (JobMaster)         (TaskManager Slots)

1. StreamGraph:
   Topology constructed directly from the user's Java/Python DataStream code.
   Represents operators (Source, Map, Window, Sink) and data streams.

2. JobGraph:
   Generated on the client by applying Operator Chaining optimizations.
   Individual operators are fused into chained execution vertices to eliminate
   inter-thread serialization.

3. ExecutionGraph:
   Constructed inside the JobMaster.
   Expands the JobGraph to match Job Parallelism. If an operator has parallelism = 4,
   the ExecutionGraph contains 4 parallel ExecutionVertex instances.

4. Physical Execution Tasks:
   The actual Java threads deployed to specific TaskManager TaskSlots.

3. Operator Chaining: Eliminating Inter-Thread Overhead

Without optimization, transferring data between two operators (e.g. Source and Map) requires: 1. Thread context switching. 2. Serializing Java objects into byte buffers. 3. Passing buffers through network queues.

Unchained Operators (Heavy Overhead):
[Source Thread] ──► [Queue Buffer] ──► [Thread Context Switch] ──► [Map Thread]

Chained Operator (Fused Zero-Copy Execution):
┌────────────────────────────────────────────────────────┐
│ Chained Task Thread (Single Thread Execution)          │
│ Source.run() ──► Direct Java Method Call ──► Map.map() │
└────────────────────────────────────────────────────────┘

3.1 Rules for Chaining Operators

Flink will automatically chain Operator A to Operator B if and only if: 1. Both operators share the same parallelism. 2. Operator B consumes data via a forward partitioner (Partitioner.FORWARD). 3. Neither operator is configured with .disableChaining(). 4. Both operators share the same Slot Sharing Group.

Benefit: Fusing operators into chains yields a 5x to 10x reduction in latency and eliminates GC serialization churn!