Developing Applications On Managed Service For Apache Flink

Job Graph Architecture Guide

Overview

This guide covers Flink job graph design for Managed Service for Apache Flink applications: operator chaining, operator-to-task-slot mapping, and task slot overload diagnosis. Use it when designing a new job graph, diagnosing performance problems in an existing one, or deciding whether to split a large application. For anti-patterns (data skew, monolith jobs, high fan-out), see job-graph-anti-patterns.md.

Operator Chaining and Job Graph Basics

How Operator Chaining Works

Flink automatically chains operators that meet all of these conditions into a single task:

• Same parallelism

• Connected by a forward (non-shuffle) data exchange

• Same slot sharing group

• Neither operator has chaining explicitly disabled

Chained operators run in the same thread, eliminating serialization/deserialization overhead and thread context switching between them. In the Flink Web UI, chained operators appear as a single box in the job graph with names joined by arrows (e.g., Source → Map → Filter).

Verifying chaining in the Flink Web UI:

1. Open the Flink Web UI from the Managed Service for Apache Flink console

2. Navigate to the running job's "Overview" tab

3. Each box in the job graph represents one task (a chain of operators)

4. Click a task box to see which operators are chained inside it

5. If operators you expected to be chained appear as separate boxes, check that parallelism matches and the data exchange is forward

Diagnosing unexpected chain breaks — work this list in order before assuming a bug or re-running the job:

1. Parallelism mismatch. Confirm every operator in the supposed chain runs at the same parallelism. Source connectors are a frequent culprit — a Kinesis source defaults to one subtask per shard, so a 4-shard stream can't chain with downstream operators set to 8. Per-operator setParallelism() overrides the env default.

2. Implicit shuffle. Any keyBy(), rebalance(), shuffle(), broadcast(), or rescale() between two operators inserts a network exchange and breaks the chain at that point. The exchange-label arrow (HASH, REBALANCE, FORWARD) between job-graph boxes tells you which case applies.

3. startNewChain() or disableChaining() left in code. These are commonly added for diagnostic isolation (see below) and forgotten.

4. Different slot sharing groups. A slotSharingGroup("...") on one operator that differs from its neighbor will prevent chaining even if everything else aligns.

Breaking a chain is not always wrong — see "Using disableChaining() and startNewChain() Strategically" below for legitimate reasons (operator-metric isolation, external-call latency separation, explicit parallelism boundaries). The Flink Web UI is the source of truth here, not "the display might be misleading."

Operator-to-Task-Slot Mapping

Each task slot runs one parallel pipeline of chained operators. The number of operators per task slot depends on how many operators chain together and how many slot sharing groups exist.

Rule of thumb: 20–40 operators per task slot. This range balances resource utilization against overhead:

Using disableChaining() and startNewChain() Strategically

Break chains only when you have a specific reason:

// Break the chain before a CPU-intensive operator to isolate its metrics
DataStream<Result> results = events
    .keyBy(Event::getKey)
    .process(new ExpensiveProcessor())
    .startNewChain()  // This operator starts a new chain
    .uid("expensive-processor-uid");

// Completely disable chaining for a specific operator (rarely needed)
DataStream<Enriched> enriched = events
    .map(new ExternalServiceLookup())
    .disableChaining()  // Runs in its own task, not chained with anything
    .uid("external-lookup-uid");

When to break chains:

• To isolate a CPU-intensive operator so its metrics (busyTime, backpressure) are visible independently in the Flink Web UI

• To separate an operator that makes external calls (HTTP, database) from the rest of the pipeline so its latency does not mask upstream/downstream metrics

• To control parallelism boundaries — operators with different parallelism cannot chain anyway, but startNewChain() makes the intent explicit

When NOT to break chains:

• For debugging only — use the Web UI's subtask metrics instead

• To "improve parallelism" — breaking chains does not change parallelism; it only adds serialization overhead

• Preemptively on every operator — this defeats the purpose of chaining and increases resource consumption

Viewing the Physical Execution Plan

The Flink Web UI shows two views of the job:

1. Job Graph (Overview tab): Shows the logical plan with chained operators grouped into task boxes. Use this to verify chaining decisions.

2. Task Managers tab: Shows which task slots are allocated on each TaskManager and what tasks run in each slot. Use this to verify operator-to-task-slot distribution.

To check operator-to-task-slot assignments:

1. Open the Flink Web UI → select the running job

2. Click on a task box in the job graph to expand its details

3. The "Subtasks" tab shows each parallel instance (subtask) and which TaskManager hosts it

4. Cross-reference with the TaskManagers tab to see total slot utilization per TaskManager

For data skew detection and mitigation, the monolith job anti-pattern, and the high fan-out anti-pattern, see job-graph-anti-patterns.md.

Operator-to-Task-Slot Overload

Recommended Ratios and Performance Implications

Each task slot runs a parallel pipeline of chained operators within a single thread. The more operators packed into a slot, the more work that thread must perform — including state access, serialization, timer management, and checkpoint barrier handling.

Recommended ratio: 20–40 operators per task slot.

Symptoms of Task Slot Overload

• High GC pressure: heapMemoryUtilization sustained above 80% (scale-up signal; see monitoring-and-metrics.md), frequent full GC pauses

• Slow checkpoints: lastCheckpointDuration increasing or approaching the checkpoint interval

• Increased latency: busyTimeMsPerSecond approaching 1000 (fully saturated)

• Unresponsive heartbeats: In extreme cases, TaskManagers miss heartbeat deadlines causing restarts

Threshold: >200 Operators → Split or Restructure

1. Split the job into independent Flink applications for processing paths that don't share state

2. Restructure the graph: combine sequential map/filter operations into a single ProcessFunction; remove redundant operators

3. Adjust slot sharing groups to distribute operators across more slots:

DataStream<Result> results = events
    .keyBy(Event::getKey)
    .process(new HeavyProcessor())
    .slotSharingGroup("heavy-processing")
    .uid("heavy-processor-uid");

Operator Group Scheduling

Flink's slot sharing allows operators from different pipeline parts to share the same task slot. When all operators are in the default group and the job has many operators, every slot runs one subtask of every operator — leading to overload.

Strategies: Group operators by resource profile (CPU-intensive vs I/O-bound in separate groups). Use the Flink Web UI's TaskManagers tab to verify balanced slot utilization.

References

• See Resource Optimization for KPU sizing, operator parallelism tuning, and checkpoint resource impact

• See Best Practices for state management and serialization guidance

• See Monitoring and Metrics for CloudWatch metric details and alarm configuration

• See Managed Service for Apache Flink Overview for KPU resource model and service constraints