DE Mastery GitHub

Build your own mini stream processor · Stage 2 of 8

Keyed partitioning across workers

Flink's keyBy routes every record for a key to the same parallel subtask (key group), so keyed state and keyed aggregation are correct regardless of parallelism. You implement that routing.

New API this stage

stream.key_by(key_fn) -> KeyedStream
keyed.reduce(reduce_fn, parallelism=1) -> KeyedResult
run_parallel(KeyedResult) -> dict          # {key: reduced_value}
partition_of(key, num_partitions) -> int   # which worker a key lands on
worker_assignment(keys, n) -> {key: worker}
stable_hash(value) -> int                  # provided; deterministic across procs

This stage's task

  • partition_of(key, n) = stable_hash(key) % n. Use the provided stable_hash (blake2b of repr), not Python's hash() (which is salted per interpreter — a test runs stable_hash in a child with a different PYTHONHASHSEED and asserts equality).
  • run_parallel drives the chain, groups data events by key, routes each key to its worker, and reduces per key with reduce_fn.
  • Because each key lives on exactly one worker, the merged result is invariant to parallelism — parallelism=1, 3, 7 all give the same answer, equal to a plain per-key fold (the oracle).
  • worker_assignment(keys, n) returns {key: partition_of(key, n)}.

Constraints

  • Deterministic routing (stable hash). No sleeps. Suite < 60 s.
  • Earlier stages still pass (regression).

Run it

de start FLK-BYO/02
de test

Hints

Hint 1

keyBy sends every record with the same key to the SAME worker. The partition of a key is a stable hash of the key modulo the worker count: partition_of(key, n) = stable_hash(key) % n. stable_hash is provided (blake2b of repr) so it is identical across processes — do NOT use Python's built-in hash(), which is salted per interpreter.

Hint 2

run_parallel drives the chain, groups data events by key, routes each key to its worker via partition_of, and reduces per key with reduce_fn. Because a key lives on exactly ONE worker, merging workers never has a key collision — so the result is identical for any parallelism (the key-group invariance the tests assert).

Hint 3

worker_assignment(keys, n) returns {key: partition_of(key, n)}. Keep reduce associative-friendly: start a key's accumulator at its first value, then reduce_fn(acc, next_value).

Run it locally
uv run de start FLK-BYO/02-keyed-partitioning
uv run de test