DE Mastery GitHub

Apache Flink · FLK-01

Philosophy & architecture

  • 4 h

From your world. the Spark driver/executors model, re-cast for always-on streaming

← All exercises

Shape a job graph and its slot demand

The job graph and the parallelism decide the resource footprint of a streaming job. You do not decide it by guessing the number of executors. In this exercise, you build a small keyed pipeline and report how many task slots it needs.

Task

Implement de.flk.JobShape.plan(int parallelism):

  1. Build a StreamExecutionEnvironment and set the job-wide parallelism to parallelism.
  2. Wire this pipeline over JobShape.sampleData() (already provided): fromData → filter(amount ≥ 0) → keyBy(key) → sum(amount) → sink. Use any sink (for example DiscardingSink). The tests do not assert the data. They assert the shape.
  3. Return a PlanReport with:
    • plan = env.getExecutionPlan() (the real JobGraph JSON),
    • jobParallelism = env.getParallelism(),
    • slotsNeeded = the number of task slots the job needs under default slot sharing.

What the tests check

  • jobParallelism equals the requested parallelism.
  • slotsNeeded equals the maximum operator parallelism (= the job parallelism here). It does not equal the sum of the parallelism of every operator. Slot sharing lets one slot hold one slice of the whole pipeline.
  • The plan contains a HASH exchange. This proves that the keyBy broke the operator chain between the filter region and the keyed aggregation.
  • The keyed-aggregation vertex runs at the requested parallelism.

Hints

  • env.getExecutionPlan() returns the JobGraph as JSON. Each node has a parallelism field. Each edge has a ship_strategy (FORWARD/REBALANCE/HASH).
  • A keyBy produces the HASH edge.
  • Under default slot sharing, slot demand is the max parallelism, not the sum. This is the main point of the exercise.

Starter code

src/main/java/de/flk/JobShape.java
package de.flk;

import org.apache.flink.api.java.tuple.Tuple2;

import java.util.ArrayList;
import java.util.List;

/**
 * FLK-01 starter — NOT implemented yet (the tests must fail: G1).
 *
 * Build a keyed streaming pipeline (source -> filter -> keyBy -> sum
 * -> sink) at the requested parallelism and return the real job graph JSON
 * plus the job-wide parallelism and the number of task slots the job needs.
 */
public class JobShape {

    public static List<Tuple2<String, Integer>> sampleData() {
        List<Tuple2<String, Integer>> data = new ArrayList<>();
        data.add(Tuple2.of("c1", 10));
        data.add(Tuple2.of("c2", 7));
        data.add(Tuple2.of("c1", 20));
        data.add(Tuple2.of("c2", 3));
        data.add(Tuple2.of("c1", -5));
        return data;
    }

    public static PlanReport plan(int parallelism) throws Exception {
        // TODO: build a StreamExecutionEnvironment, set parallelism, wire
        //       fromData -> filter -> keyBy -> sum -> sink, then return
        //       env.getExecutionPlan(), env.getParallelism() and the slot demand.
        return new PlanReport("{\"nodes\":[]}", 0, 0);
    }

    public static final class PlanReport {
        public final String plan;
        public final int jobParallelism;
        public final int slotsNeeded;

        public PlanReport(String plan, int jobParallelism, int slotsNeeded) {
            this.plan = plan;
            this.jobParallelism = jobParallelism;
            this.slotsNeeded = slotsNeeded;
        }
    }
}

Hints

Hint 1

Build a StreamExecutionEnvironment, set the job-wide parallelism to the requested value, and read it back from env.getParallelism().

Hint 2

A chain is broken by a keyBy. Build source -> keyBy(...) -> sum/reduce -> sink so the plan has two chained regions (an exchange between them).

Hint 3

Slot demand under default slot sharing equals the MAX operator parallelism, so return that number for slots_needed; keep every operator at the job parallelism.

Run it locally
uv run de start FLK-01/01-job-graph-slots
# edit workspace/FLK-01/01-job-graph-slots/
uv run de test