How do you change the parallelism of a running Flink job, and what caps it?

## Why it is a restart Under Flink's default scheduler the ExecutionGraph is built once, with a fixed parallelism per operator, and subtasks are deployed into slots for the life of the run. There is no in-place resize. Changing parallelism therefore means: snapshot the state, stop, and start a new run that restores from the snapshot at the new parallelism. ## The procedure ```bash # 1. snapshot and stop cleanly ./bin/flink stop --savepointPath s3://bucket/savepoints <jobId> # 2. restore at the new parallelism ./bin/flink run -s s3://bucket/savepoints/savepoint-xxxx -p 16 job.jar ``` `stop` with a savepoint is preferable to `cancel` because it stops the sources first and takes a final, consistent snapshot, so no work is repeated. Do **not** pass `--drain` when you intend to resume: draining emits a final maximum watermark, firing every remaining window and advancing event time to infinity, which corrupts the semantics of the resumed job. A retained checkpoint can also be used as a restore point when no savepoint exists, but the savepoint is the deliberate, portable one. ## What actually limits the new parallelism Keyed state is not partitioned by subtask directly — it is partitioned by **key group**. Flink hashes each key into one of *N* key groups, and each subtask owns a contiguous range of them. That indirection is what makes rescaling possible: on restore, the same *N* key groups are simply divided among a different number of subtasks, and each subtask reads exactly the key-group ranges it now owns. *N* is the operator's **maximum parallelism** (`setMaxParallelism(...)` or `pipeline.max-parallelism`), and it is fixed in the savepoint. Two consequences: - **Parallelism can never exceed maximum parallelism.** Ask for more and the job fails to start. - **Maximum parallelism cannot be changed on restore.** Changing it changes which key goes to which key group, so the restored state would be wrong; Flink refuses. Escaping the ceiling means rewriting the savepoint offline with the State Processor API — an expensive, rarely-run operation. If you never set it, Flink derives a maximum parallelism from the initial parallelism, choosing a value of at least 128 and at most 32768. A job first launched at parallelism 4 therefore usually gets 128, which is fine until the day you need 200 subtasks. Setting it explicitly at job creation — high enough to be a real ceiling, not so high that per-key-group overhead and metadata grow needlessly — is the cheap decision that avoids the expensive one. It is also worth choosing a parallelism that divides the key-group count evenly. With 128 key groups and parallelism 12, some subtasks own 11 groups and others 10, an ~10% built-in imbalance; parallelism 8, 16 or 32 divides cleanly. ## Non-keyed state and operator identity Operator (non-keyed) state is redistributed on restore too, either by splitting the list evenly across the new subtasks or by giving every subtask the union of all entries, depending on how the operator declared it. Kafka source partition offsets are the everyday example of the even-split form. All of this depends on Flink being able to match state in the savepoint to operators in the new job. It does that by operator ID, which is generated from the graph structure unless you assign one with `uid("...")`. Without explicit UIDs, adding or removing an operator can shift the generated IDs and the restore fails or silently drops state. Assign a stable `uid()` to every stateful operator from the first deployment; add `--allowNonRestoredState` only when you have deliberately removed an operator and accept discarding its state. ## The resource side A higher parallelism needs more slots — the requirement is the maximum operator parallelism under default slot sharing. On native Kubernetes or YARN, Flink's ResourceManager will request more TaskManager containers; in a standalone cluster you must add TaskManagers yourself, or the job sits in scheduling and times out. Raising parallelism also raises the source ceiling problem: a Kafka source cannot usefully run wider than the topic's partition count, so scaling the job without repartitioning the topic leaves idle source subtasks. ## Cost of the operation The rescale is a full stop and restart. Downtime is savepoint duration plus job startup plus state reload — and with a large embedded RocksDB state, reload means downloading and rebuilding many gigabytes per subtask, which can dominate. That cost is the practical reason teams size long-running jobs for peak and rescale rarely, rather than tracking load closely.