Performance & state at scale — backpressure, parallelism, redistribution, skew

6 h
requires FLK-02
requires FLK-06

From your world. Spark skew and shuffle tuning: one busy task holding up the stage, fixed by redistributing work across more parallel subtasks

← All exercises

Fix a backpressured job — raise parallelism and redistribute

A streaming job runs a CPU-heavy map over a source. The baseline runs that map at parallelism 1 with a FORWARD edge. One subtask does all the work and backpressures the source. The whole job runs at the speed of that one subtask. Your task is to fix the bottleneck. The work of the heavy map must spread across several parallel subtasks. The result must not change.

The provided computation (do not change its logic)

de.flk.Work.busy(long) is the per-record CPU work (it comes with the exercise). Both the baseline and your version call the same Work::busy, so the comparison is fair.

What to implement

Edit de.flk.Fix.apply(DataStream<Long> source) to return the heavy-mapped stream, optimized:

public static DataStream<Long> apply(DataStream<Long> source) {
    // TODO: redistribute the records, then run the heavy map on more subtasks.
    // return source.rebalance()
    //              .map((MapFunction<Long,Long>) Work::busy)
    //              .name("heavy")
    //              .setParallelism(4);
}

Requirements:

The heavy map must keep the name "heavy" and call Work::busy (unchanged result).
Run the heavy map at parallelism > 1.
Insert a redistributing edge before it, so the extra subtasks get records: rebalance() (even round-robin) or rescale(). Do not keyBy a constant. That sends every record to a single subtask, which is the bottleneck that you want to remove.

How you are graded (robust, not a stopwatch)

The test builds the baseline itself and compares its job graph to yours:

The plan changed in the right direction. The parallelism of your "heavy" operator increased over the baseline. The edge that feeds it is a redistributing strategy (REBALANCE / RESCALE) and not FORWARD. The test rejects a keyBy (HASH) edge. The heavy map is not keyed, so hashing only brings back skew. This is the deterministic proof that throughput improves: more subtasks do balanced work.
Correctness preserved. The summed output of your job equals the baseline (and an independent DuckDB oracle over Work::busy).

The test grades nothing on wall-clock time, so results are stable even on a busy machine.

Starter code

src/main/java/de/flk/Fix.java

package de.flk;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;

/** Starter — this is the bottleneck itself: the heavy map runs on a single
 *  subtask with a FORWARD edge. Fix it (see README): more parallelism + a
 *  redistributing edge, same result. */
public class Fix {
    public static DataStream<Long> apply(DataStream<Long> source) {
        // TODO: redistribute and raise parallelism.
        return source
            .map((MapFunction<Long, Long>) Work::busy)
            .name("heavy")
            .setParallelism(1);
    }
}

src/main/java/de/flk/Work.java

package de.flk;

/** Provided CPU-heavy per-record computation. Do not change its logic — both
 *  the baseline and your Fix call Work::busy, so the comparison is fair and the
 *  result (the summed output) must be identical. */
public final class Work {
    private Work() {}

    public static long busy(long x) {
        long s = 0;
        for (int i = 0; i < 2000; i++) {
            s += (x * 2654435761L + i) % 97;
        }
        return s % 1000;
    }
}

Hints

Hint 1

The baseline runs the heavy map at parallelism 1 with a FORWARD edge, so one subtask does all the work. Give the map more subtasks with .setParallelism(n) (n>1).

Hint 2

Parallelism alone is not enough — a FORWARD edge can't feed more subtasks. Insert a redistributing edge before the map: source.rebalance().map(...). rescale() also works.

Hint 3

Keep the operator named "heavy" (.name("heavy")) and keep the computation Work::busy so the result is unchanged. Do NOT keyBy a constant — that funnels everything to one subtask.

Run it locally

uv run de start FLK-11/01-fix-backpressure
# edit workspace/FLK-11/01-fix-backpressure/
uv run de test