Parallelism, Task Slots, and Operator Chaining
Every operator in a Flink job has a parallelism — the number of parallel instances (subtasks) it runs as — and each TaskManager offers a fixed number of task slots, each capable of running one slice of the job's pipeline. Flink chains adjacent operators with compatible parallelism and forwarding partitioning (like map followed by filter) into a single task to avoid serialization and network overhead, so what looks like five operators in your code might execute as a single fused task per slot, only splitting into separate tasks at shuffle boundaries like keyBy.
Cricket analogy: It's like a fielding team where the slip cordon (chained operators: first slip, second slip, gully) all react as one coordinated unit to a nick, while a shuffle boundary is like the ball actually being thrown to a different fielder across the ground — that throw has real overhead the adjacent slips don't.
Setting Parallelism at Different Levels
Parallelism can be set at four levels, from broadest to narrowest: the cluster default (parallelism.default in flink-conf.yaml), the job level via execution.parallelism or env.setParallelism(n), and the operator level via .setParallelism(n) called on an individual DataStream transformation, with operator-level settings always overriding job and cluster defaults. A common pattern is a low default parallelism with a higher parallelism explicitly set on the specific operator known to be the bottleneck, such as a CPU-heavy enrichment map following a lightweight Kafka source.
Cricket analogy: It's like setting a default number of fielders in the deep for the whole innings, but overriding it for a specific over against a known big hitter — the general plan (cluster default) gives way to a targeted field placement (operator-level override).
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4); // job-level default
DataStream<Order> orders = env.fromSource(kafkaSource, watermarkStrategy, "orders")
.setParallelism(2); // narrow source parallelism to match Kafka partition count
DataStream<EnrichedOrder> enriched = orders
.map(new ExpensiveEnrichmentFunction())
.setParallelism(16); // scale up the CPU-heavy bottleneck operator
enriched.keyBy(o -> o.getCustomerId())
.process(new FraudDetectionFunction())
.setParallelism(8)
.sinkTo(kafkaSink);Rescaling with Savepoints
Changing parallelism on a running job requires stopping it with a savepoint (flink stop --savepointPath s3://bucket/savepoints job-id) and restarting from that savepoint with a new parallelism; Flink redistributes keyed state across the new number of subtasks using key groups — a fixed number (default 128, set via max-parallelism) of hash buckets that state is pre-partitioned into, so rescaling reshuffles key groups across subtasks rather than requiring a full state migration per key. Because max-parallelism caps how far a job can ever scale, it should be set generously upfront (commonly 128-32768) since changing it requires a full state migration rather than a simple rescale.
Cricket analogy: It's like a franchise's fixed pool of 128 registered squad numbers — reshuffling which 11 players are picked for a match (rescaling) is easy within that pool, but expanding the pool itself mid-season requires re-registering everyone from scratch.
Flink 1.13+ supports Reactive Scaling Mode (adaptive scheduler), where the job automatically adjusts parallelism up to max-parallelism when TaskManagers join or leave the cluster — useful for autoscaling on Kubernetes without manually triggering savepoint-based rescales.
Setting parallelism higher than the number of Kafka partitions on a source operator wastes resources — extra source subtasks sit idle since a partition can only be consumed by one subtask at a time within a consumer group.
- Parallelism defines how many parallel subtasks an operator runs as; task slots are the physical execution containers on TaskManagers.
- Flink chains compatible adjacent operators into single tasks to avoid unnecessary serialization and network overhead.
- Parallelism can be set at cluster, job, and operator level, with operator-level settings taking precedence.
- Rescaling a running job requires stopping with a savepoint and restarting with new parallelism.
- Key groups (bounded by max-parallelism) pre-partition keyed state so rescaling redistributes groups rather than migrating per-key.
- max-parallelism should be set generously upfront since changing it later requires a full state migration.
- Reactive Scaling Mode enables automatic parallelism adjustment for autoscaling deployments, especially on Kubernetes.
Practice what you learned
1. What does operator chaining primarily eliminate between compatible adjacent operators?
2. Which level of parallelism configuration takes precedence when multiple levels are set?
3. What mechanism allows Flink to redistribute keyed state efficiently when rescaling?
4. Why is it costly to increase max-parallelism after a job has been running in production?
5. What happens if a Kafka source operator's parallelism exceeds the number of partitions in its topic?
Was this page helpful?
You May Also Like
Flink with Kafka
Understand how to build reliable, exactly-once streaming pipelines by connecting Apache Flink to Apache Kafka as both source and sink.
Monitoring with the Flink Dashboard
Learn to read Flink's web dashboard and metrics system to diagnose backpressure, checkpoint failures, and skewed workloads.
Deploying Flink on Kubernetes
Understand the native Kubernetes deployment modes for Flink, how the Flink Kubernetes Operator manages job lifecycle, and key production configuration.
Related Reading
Related Study Notes in Programming
Browse all study notesApache Spark Study Notes
Programming · 30 topics
ProgrammingHadoop Study Notes
Programming · 30 topics
ProgrammingSnowflake Study Notes
Programming · 30 topics
ProgrammingApache Airflow Study Notes
Programming · 30 topics
Programmingdbt (Data Build Tool) Study Notes
Programming · 30 topics
ProgrammingPowerShell Study Notes
Programming · 30 topics