Learn Apache Flink - Scaling & Resource Management

#Introduction

Episode 13 taught you how to observe the system. Episode 14 uses that data to take action: adding or reducing resources. Scale in Flink isn't automatic — you have to understand how to tune parallelism, slot sizes, and state backends so your pipeline can keep up with the load.

We'll cover parallelism and task slot sizing, autoscaling for Flink on Kubernetes, controlling backpressure and throughput, and a deep comparison of the RocksDB versus in-memory state backends. This is the episode where tuning theory starts to coalesce into practice.

#Parallelism and Task Slot Sizing

#Basic Slot Rules

The number of slots a job needs equals its parallelism. Two common models:

• Large slots per TaskManager: one slot per large subtask — suitable for heavy stateful jobs.
• Many small slots per TaskManager: small subtasks sharing a process — suitable for lightweight, resource-efficient jobs.

DataStream<Order> orders = source
    .map(new ParseOrder()).setParallelism(8)
    .keyBy(Order::getUserId)
    .process(new Aggregator()).setParallelism(4)
    .sinkTo(sink).setParallelism(2);

setParallelism gives different parallelism per operator according to the load. Light parsing can run 8 subtasks, stateful aggregation is fine with 4, and the sink is limited to 2 so it doesn't overload the destination database.

#Measuring the Right Slots

Start with the CPU and memory per TaskManager, then divide by the number of slots. If one TaskManager has 4 CPUs and 8GB of memory, four slots with 2GB each is a reasonable starting point. Watch the dashboard: if all slots are full and backpressure is high, raise parallelism; if idle for long, lower it.

#Autoscaling Flink on Kubernetes

#Reactive Mode

Flink supports reactive mode for automatic scaling based on parallelism: when the number of TaskManagers changes, the job adjusts its parallelism without a restart. Enable it in config.yaml:

jobmanager.scheduler: reactive
parallelism.default: 2
taskmanager.numberOfTaskSlots: 1

jobmanager.scheduler: reactive makes the JobManager recompute parallelism from the number of available slots. With slots set to 1, adding a TaskManager automatically raises parallelism — the foundation of autoscaling on Kubernetes.

#Combining with HPA

On Kubernetes, combine reactive mode with a Horizontal Pod Autoscaler that watches metrics like CPU or throughput. When load rises, the HPA adds TaskManager replicas; reactive mode absorbs the new parallelism without stopping the job. This is the most practical autoscaling pattern for production Flink.

Scaling doesn't always mean adding. When load drops — for example at night for e-commerce — the HPA reduces TaskManager replicas and reactive mode lowers parallelism. The job never stops; it shrinks first, then grows again when traffic recovers. Make sure the metrics the HPA uses reflect real load, not just CPU, which can mislead on jobs that are waiting for data.

#Controlling Backpressure and Throughput

#Tuning Network Buffers and Latency

Backpressure often comes from network buffers that are too small. Buffer size and latency bounds are configured in the config:

taskmanager.memory.network.min: 64mb
taskmanager.memory.network.max: 128mb
execution.buffer-timeout: 100ms

execution.buffer-timeout controls how quickly records are sent between operators. A small value lowers latency but raises overhead; a large value increases throughput at the cost of latency. Find the sweet spot that fits your needs.

#Reading Throughput

Throughput is measured from the numRecordsInPerSecond and numRecordsOutPerSecond metrics per operator. Falling output while input is normal is the signature of a bottleneck. After fixing the operator (higher parallelism, optimized query), compare before and after numbers to verify the improvement.

curl -s "http://localhost:8081/jobs/<jobId>/metrics?get=numRecordsOutPerSecond"

The curl -s .../metrics?get=numRecordsOutPerSecond command reads a specific metric value directly — much faster than guessing from the dashboard.

#State Backend Optimization

#RocksDB vs Memory

The state backend choice affects your scaling limits:

• HashMap (memory): state in the heap — very fast, but limited by memory size and prone to GC.
• RocksDB: state on local disk with an in-memory cache — can store huge state, slightly slower.

state.backend.type: rocksdb
state.backend.incremental: true
state.backend.rocksdb.memory.managed: true
taskmanager.memory.managed.fraction: 0.6

taskmanager.memory.managed.fraction determines the memory portion allocated to state. A value of 0.6 means 60 percent of TaskManager memory goes to RocksDB. For small state, the hashmap backend is simpler; for tens of gigabytes of state, RocksDB is the only realistic choice.

#Practices for Reducing State

State size is a hidden cost in every job. Reduce it by: using MapState instead of ValueState for multiple entities, setting state TTL (covered in episode 15), and designing keys with reasonable granularity. Slim state means faster checkpoints and easier scaling.

#Conclusion

Episode 14 equipped you with resource control: tuning parallelism and slots per operator, applying reactive mode with HPA for Kubernetes autoscaling, controlling backpressure through buffers and timeouts, and choosing and optimizing the RocksDB versus in-memory state backends.

The key takeaways:

• Total slots must cover parallelism; set parallelism per operator according to load.
• Reactive mode adjusts parallelism from the slot count without a job restart.
• execution.buffer-timeout balances latency and throughput.
• RocksDB for large state, hashmap for small state; adjust the managed fraction.
• Slim state makes checkpoints and scaling more efficient.

In the next episode, episode 15, we'll discuss performance tuning — optimizing the job graph and operator chaining, tuning checkpoint intervals and state size, garbage collection tuning, and measuring throughput and end-to-end latency. This episode turns a running job into a fast job.