Back to Apache Flink
Q33Question 33 of 33

What is the relationship between parallelism and task slots in Flink

Key Concepts
  • Parallelism is the number of parallel subtasks an operator is split into, and each subtask needs exactly one task slot to execute.
  • A task slot is a unit of execution resource on a TaskManager representing a reserved share of its managed memory.
  • The maximum parallelism a job can run is capped by the total number of slots across all TaskManagers in the cluster.
  • Slot sharing lets subtasks from different operators share one slot, so a pipeline needs only as many slots as its highest operator parallelism.
  • Default slot sharing groups all operators together, which improves resource utilization and load balance by mixing heavy and light operators.
  • You can isolate resource-hungry operators into separate slot sharing groups so they get dedicated slots.
  • Operator chaining fuses consecutive forwarding operators into a single task thread within a slot, eliminating serialization overhead.

Definition

Parallelism is subtasks per operator and each subtask needs a slot, so parallelism is capped by total slots, and slot sharing lets different operators' subtasks share one slot so a pipeline needs only as many slots as its highest parallelism.
  • Parallelism – The number of parallel subtasks an operator is split into, where each subtask processes a slice of the data stream independently.
  • Task slot – A unit of execution resource on a TaskManager that represents a reserved share of its managed memory and can host one or more subtasks.
  • Slot constraint – The job's maximum parallelism cannot exceed the total number of slots available across all TaskManagers.
  • Slot sharing – A mechanism that lets subtasks from different operators of the same job run together in one slot, reducing the total slots needed to the highest operator parallelism.
  • Slot sharing group – A named group that controls which operators may share slots, letting you isolate resource-hungry operators into dedicated slots.

Imagine the National Bank of India processing fraud checks on a conveyor line with several stations: receive the transaction, look up the account, score the fraud risk, and raise an alert. Parallelism is how many copies of the conveyor line you run side by side, say four lines to handle four transactions at once. A task slot is a desk where a worker can sit. Slot sharing is the clever arrangement where one worker at a single desk handles all four stations for one transaction in sequence, so you only need four desks, one per line, instead of sixteen. Fewer desks, same throughput.

Understanding the Concepts

When you set an operator's parallelism to four, Flink splits that operator into four independent subtasks. Each subtask processes a disjoint slice of the data stream in parallel. You can set parallelism at the job level, per operator, or via the cluster configuration.

JAVA
package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class ParallelismDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        // Job-wide default parallelism
        env.setParallelism(4);

        env.addSource(new KafkaSource<>("nbi-transactions"))
            .setParallelism(8)  // override: source reads 8 Kafka partitions
            .keyBy(Transaction::getAccountId)
            .process(new FraudScorer())
            .setParallelism(4)  // 4 subtasks for scoring
            .addSink(new AlertSink())
            .setParallelism(2); // 2 subtasks for alerting

        env.execute("NBI Parallelism Demo");
    }
}

What are task slots?

A task slot is the runtime unit of scheduled resource on a TaskManager. Each TaskManager has a configurable number of slots, and each slot represents a reserved share of the TaskManager's managed memory. The total slots in the cluster equals slots-per-TaskManager multiplied by the number of TaskManagers.

TEXT
# flink-conf.yaml
taskmanager.numberOfTaskSlots: 4    # 4 slots per TaskManager
# With 3 TaskManagers: 4 x 3 = 12 total slots
JAVA
package com.bank.streaming;

public class SlotCalculation {
    public static void main(String[] args) {
        int slotsPerTM = 4;
        int taskManagers = 3;
        int totalSlots = slotsPerTM * taskManagers; // 12
        int maxParallelism = 8; // highest operator parallelism

        System.out.println("Total slots: " + totalSlots);
        System.out.println("Max parallelism: " + maxParallelism);
        System.out.println("Can deploy: " + (totalSlots >= maxParallelism));
        // Output: Total slots: 12, Max parallelism: 8, Can deploy: true
    }
}

Every parallel subtask must be placed into a task slot to execute. If you set parallelism to 8 but the cluster has only 4 slots, the job cannot be fully deployed and Flink either waits for slots or fails. The fundamental constraint is that parallelism must be less than or equal to total slots.

JAVA
package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SlotConstraintDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        // Cluster has 4 slots but we request parallelism 8
        env.setParallelism(8);

        env.fromElements("ACC-001:5000", "ACC-002:3000")
            .map(line -> line.toUpperCase())
            .print();

        // This will fail: "Not enough slots available"
        // org.apache.flink.runtime.jobmanager.scheduler
        //   .NoResourceAvailableException
        env.execute("Slot Constraint Demo");
    }
}
TEXT
Exception: Not enough slots available to run the job.
Required: 8, Available: 4

What is slot sharing and how does it work?

Without slot sharing, a pipeline of four operators each at parallelism four would need sixteen slots. Slot sharing lets one subtask from each operator share a single slot, so the pipeline needs only four slots. By default all subtasks of a job belong to the same slot sharing group.

JAVA
package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SlotSharingDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // 4 operators, parallelism 4 each
        // Without slot sharing: 4 x 4 = 16 slots needed
        // With slot sharing (default): only 4 slots needed
        env.addSource(new KafkaSource<>("nbi-transactions"))  // operator 1
            .map(Transaction::enrich)                          // operator 2
            .keyBy(Transaction::getAccountId)                  // operator 3
            .process(new FraudScorer())                        // operator 4
            .print();

        env.execute("NBI Slot Sharing Demo");
        // Each slot runs: Source -> Map -> KeyBy -> FraudScorer
    }
}

How is slot sharing different from operator chaining?

| Feature | Slot Sharing | Operator Chaining |
|---|---|---|
| What it does | Multiple subtasks from different operators share one slot | Consecutive operators fused into one task thread |
| When it applies | Default for all operators in a job | Only when operators have same parallelism and no shuffle |
| Network cost | Subtasks in same slot may still exchange via network | Chained operators have zero serialization or network |
| Control | slotSharingGroup("name") | disableChaining() or startNewChain() |
| Benefit | Fewer slots needed, better utilization | Lower CPU and latency within a chain |

JAVA
package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class ChainingVsSharingDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // Source -> Filter -> Map are CHAINED (same parallelism, no shuffle)
        // They run in one thread with zero serialization
        env.addSource(new KafkaSource<>("nbi-transactions"))
            .filter(txn -> txn.getAmount().compareTo(BigDecimal.ZERO) > 0)
            .map(Transaction::enrich)
            // KeyBy forces a shuffle: breaks the chain
            .keyBy(Transaction::getAccountId)
            .process(new FraudScorer())
            .print();

        // All operators SHARE slots (default slot sharing group)
        // So 4 slots serve the entire pipeline
        env.execute("NBI Chaining vs Sharing Demo");
    }
}

How do you isolate operators with slot sharing groups?

When a particular operator is resource-hungry, you assign it to a separate slot sharing group so it gets dedicated slots. This increases the total slot requirement but prevents the heavy operator from starving lighter ones.

JAVA
package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SlotSharingGroupDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        env.addSource(new KafkaSource<>("nbi-transactions"))
            .slotSharingGroup("default")    // shares with others
            .keyBy(Transaction::getAccountId)
            .process(new HeavyMLFraudScorer())
            .slotSharingGroup("ml-scoring") // dedicated slots
            .addSink(new AlertSink())
            .slotSharingGroup("default");   // back to shared

        // Total slots needed: 4 (default group) + 4 (ml-scoring) = 8
        env.execute("NBI Slot Sharing Group Demo");
    }
}

What are common mistakes with parallelism and slots?

The most common mistake is setting parallelism higher than the number of Kafka partitions for a source, which leaves some subtasks idle. Another mistake is assuming that increasing parallelism always improves performance when the bottleneck is a slow external call or key skew.

JAVA
// WRONG: parallelism 8 but only 4 Kafka partitions
// 4 subtasks read data, 4 subtasks sit idle
env.addSource(new KafkaSource<>("nbi-transactions")) // 4 partitions
    .setParallelism(8); // waste: 4 subtasks have no data

// CORRECT: match parallelism to partition count
env.addSource(new KafkaSource<>("nbi-transactions")) // 4 partitions
    .setParallelism(4); // each subtask reads one partition

// WRONG: high parallelism with skewed keys
// keyBy(accountId) where 80% of traffic is one corporate account
// One subtask gets 80% of data, others are nearly idle
stream.keyBy(Transaction::getAccountId) // skewed!
    .setParallelism(16); // 15 subtasks underutilized

When should you adjust parallelism?

Increase parallelism when throughput is insufficient and the bottleneck operator has room to scale. Decrease it when subtasks are idle or when you are wasting slots. For keyed operators, parallelism beyond the number of distinct keys gives no benefit. Always monitor per-subtask throughput to verify that increased parallelism actually distributes load.

JAVA
package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class ParallelismTuningDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        // Source: match Kafka partitions
        env.addSource(new KafkaSource<>("nbi-transactions"))
            .setParallelism(8) // 8 Kafka partitions
            .keyBy(Transaction::getAccountId)
            .process(new FraudScorer())
            .setParallelism(4) // 4 is enough: scoring is not the bottleneck
            .addSink(new DatabaseSink())
            .setParallelism(2); // DB can only handle 2 concurrent writers

        env.execute("NBI Tuned Parallelism");
    }
}

What should a senior engineer know about parallelism and slots?

A senior engineer should understand that the right parallelism is driven by throughput requirements and key cardinality, that slot sharing is the reason a pipeline of N operators needs only max-parallelism slots instead of sum-parallelism slots, and that operator chaining is a separate optimization that fuses forwarding operators into one thread. They should also know that changing parallelism between savepoint restores requires configuring max parallelism in advance.

JAVA
package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class MaxParallelismDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // Set max parallelism upfront so savepoints remain compatible
        // when you later scale from 4 to 8
        env.setMaxParallelism(128);

        env.addSource(new KafkaSource<>("nbi-transactions"))
            .keyBy(Transaction::getAccountId)
            .process(new FraudScorer())
            .addSink(new AlertSink());

        env.execute("NBI Max Parallelism Demo");
    }
}

Visual Overview

DIAGRAM
         Parallelism and Task Slots

  Cluster: 3 TaskManagers x 4 slots = 12 total slots

  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
  │TaskManager-1│ │TaskManager-2│ │TaskManager-3│
  │┌───┐┌───┐  │ │┌───┐┌───┐  │ │┌───┐┌───┐  │
  ││S1 ││S2 │  │ ││S5 ││S6 │  │ ││S9 ││S10│  │
  │└───┘└───┘  │ │└───┘└───┘  │ │└───┘└───┘  │
  │┌───┐┌───┐  │ │┌───┐┌───┐  │ │┌───┐┌───┐  │
  ││S3 ││S4 │  │ ││S7 ││S8 │  │ ││S11││S12│  │
  │└───┘└───┘  │ │└───┘└───┘  │ │└───┘└───┘  │
  └─────────────┘ └─────────────┘ └─────────────┘


  Without Slot Sharing (4 operators x parallelism 4 = 16 slots)
  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐
  │Src-0 │ │Src-1 │ │Src-2 │ │Src-3 │  4 slots
  │Map-0 │ │Map-1 │ │Map-2 │ │Map-3 │  4 slots
  │Key-0 │ │Key-1 │ │Key-2 │ │Key-3 │  4 slots
  │Sink-0│ │Sink-1│ │Sink-2│ │Sink-3│  4 slots
  └──────┘ └──────┘ └──────┘ └──────┘  = 16 slots


  With Slot Sharing (max parallelism = 4 slots)
  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
  │ Slot 0   │ │ Slot 1   │ │ Slot 2   │ │ Slot 3   │
  │ Src-0    │ │ Src-1    │ │ Src-2    │ │ Src-3    │
  │ Map-0    │ │ Map-1    │ │ Map-2    │ │ Map-3    │
  │ Key-0    │ │ Key-1    │ │ Key-2    │ │ Key-3    │
  │ Sink-0   │ │ Sink-1   │ │ Sink-2   │ │ Sink-3   │
  └──────────┘ └──────────┘ └──────────┘ └──────────┘
              = 4 slots (75% savings)

Key Takeaway

In Flink, parallelism is the number of parallel subtasks an operator is split into, and a task slot is the runtime's unit of execution resource on a TaskManager, representing a reserved share of its managed memory. Every parallel subtask needs a slot to run, so the fundamental relationship is that the parallelism you can actually run is capped by the total number of slots in the cluster, which is the slots-per-TaskManager setting times the number of TaskManagers.

Slot sharing addresses the efficiency problem. Without it, a pipeline of four operators each at parallelism four would need sixteen slots. With default slot sharing, one subtask from each operator shares a single slot, so a slot runs the source, keyBy, scoring, and sink subtasks of the same parallel pipeline instance. The result is that the slots a job needs equal the maximum parallelism among its operators, not the sum. This also improves load balance because mixing one of each operator per slot spreads an expensive operator's work evenly instead of concentrating it.

At the National Bank of India our fraud pipeline had four operators at parallelism four, and the team initially provisioned sixteen slots assuming each subtask needed its own slot. Once we understood default slot sharing, we saw it needed only four slots, each running a complete source-to-sink slice, which let us run the same throughput on a much smaller cluster. When the ML scoring step became extremely CPU-heavy, we isolated it into its own slot sharing group for dedicated slots, understanding that this raised the total slot requirement from four to eight.

The rules I apply are to size parallelism to throughput and key cardinality, ensure total slots at least equal the maximum parallelism, rely on default slot sharing for efficiency, and isolate only the operators that genuinely need dedicated resources.

Remember one sentence: Slot sharing lets a full pipeline share one slot per parallel instance so you need only max-parallelism slots, not sum-of-all-operator subtasks.

---

Share
Use ← → arrow keys to navigate