In Flink, which parallelism setting wins — parallelism.default, env.setParallelism(), or an operator's own?

## Parallelism, concretely A Flink job is a graph of tasks — transformations, sources, sinks. Each task is split into parallel instances, and the number of instances is that task's *parallelism*. Every instance processes a subset of the task's input. Parallelism is therefore the main throughput knob, and it is set independently per operator if you want. ## The four levels **Operator level.** Call `setParallelism()` directly on the returned stream, right after the transformation it refers to: ```java DataStream<Tuple2<String, Integer>> counts = text .flatMap(new LineSplitter()) .keyBy(value -> value.f0) .window(TumblingEventTimeWindows.of(Duration.ofSeconds(5))) .sum(1).setParallelism(5); ``` This is the most specific setting and it wins for that operator alone. **Execution-environment level.** `env.setParallelism(3)` sets the default for every operator, source and sink the environment executes. Any operator that sets its own value overrides this. **Client level.** The CLI accepts `-p`: `./bin/flink run -p 10 job.jar`. This supplies the job's parallelism at submission. **System level.** `parallelism.default` in the Flink configuration file gives a cluster-wide default for all execution environments. The resolution order runs from the most specific to the most general: operator, then environment, then client, then system. In practice this is why a job submitted with `-p 32` can still run part of its graph at parallelism 5 — an explicit `setParallelism(5)` in the code is more specific and stands. That specificity ordering also explains a common operational complaint: an operator team tries to scale a job by raising `-p` at submission, and nothing changes, because the program itself calls `env.setParallelism(...)`. Leaving parallelism out of the application code is the usual fix, for the same reason the docs prefer choosing the runtime mode at submission — a configuration-free artifact is one a platform can tune without a rebuild. ## Why some operators should differ from the job default Uniform parallelism is rarely optimal. A source is often capped by the external system: a Kafka source cannot usefully run more subtasks than the topic has partitions, and extra subtasks simply idle. A sink writing to a database may need *lower* parallelism than the compute stages so it does not exhaust the connection pool or trip a rate limit. A CPU-heavy enrichment step in the middle may want more than either. Setting these individually is what the operator-level knob is for. The other side of that coin: every parallelism change between adjacent operators forces a redistribution and breaks the operator chain, so records get serialized and shipped between subtasks. A chain of `map().filter().map()` at the same parallelism runs in a single thread with no serialization; inserting one differing `setParallelism()` in the middle turns that into two tasks and a network exchange. Vary parallelism where the bottleneck actually justifies it, not by reflex. ## Maximum parallelism is a different thing `setMaxParallelism()` looks like a sibling but plays a different role. Flink partitions keyed state into *key groups*; the number of key groups equals the operator's maximum parallelism, and each subtask owns a contiguous range of them. That indirection is what lets you restore a savepoint at a different parallelism — key-group ranges are reassigned, and each subtask reconstructs its state from the groups it now owns. So max parallelism is the hard ceiling on how far a keyed operator can ever be scaled out for the life of its state. It can be set on an operator or on the environment with `setMaxParallelism()`, and in Flink 2.3 also program-wide without touching code, through the `pipeline.max-parallelism` option — in `config.yaml` or as `-Dpipeline.max-parallelism=...` at submission. There is no dedicated CLI flag like `-p`, and the parallel-execution docs page still says client and system level are excluded; the option is what the runtime actually reads. Its default is roughly `operatorParallelism + (operatorParallelism / 2)`, with a lower bound of 128 and an upper bound of 32768. Two warnings come with it. Setting it very large is not free: some state backends keep internal structures that scale with the key-group count, so an unnecessarily huge value costs performance. And changing it explicitly when recovering a job from its original state leads to state incompatibility — the key-group mapping no longer matches what the savepoint recorded. That is why a long-lived job's max parallelism is a decision made deliberately at design time, not adjusted later during an incident. ## Reading the result The web UI shows each job vertex with its parallelism, which is the fastest way to confirm what the resolution order actually produced — more reliable than reading the submission command, because the code may have overridden it. Naming operators with `.name("...")` makes that view legible. If one vertex shows high backpressure while its neighbours are idle, the parallelism of that vertex, or the key distribution feeding it, is the first thing to check.