DE Mastery GitHub

Apache Spark · SPK-10

Structured Streaming I — the micro-batch engine and exactly-once

From your world. dbt incremental models and their state

← All exercises

Pick the right output mode for a streaming aggregation

An event stream lands as JSON files. You must maintain a running per-company event count in an in-memory table that a dashboard reads. The query is an aggregation, so the output mode matters:

  • append is illegal here. The counts keep changing, so nothing is ever "final" to append.
  • complete emits the whole current result in each batch.
  • update emits only the changed rows.

Implement run_counts(spark, source_dir, query_name) in solution.py. It must do these things:

  • Read source_dir as a streaming JSON source with schema event_id INT, company_id STRING, amount DOUBLE, event_ts STRING.
  • Aggregate to one row per company_id with column n = number of events.
  • Write to the in-memory sink (format("memory")) under query_name. Choose an output mode that is valid for an aggregation and that leaves the complete current result available to query as SELECT * FROM <query_name>.
  • Use trigger(availableNow=True), call processAllAvailable(), then call stop(). Return the query object.

The test writes several batches and runs your query. Then it reads SELECT company_id, n FROM <query_name> and compares the final counts to a DuckDB oracle. A version that uses append must fail (Spark rejects it on this aggregation).

Starter code

solution.py
"""SPK-10 / 02-output-modes — starter. See README.md.

Aggregate the file stream to per-company counts and write to the memory sink under
query_name with a VALID output mode for an aggregation. Return the stopped query.
"""

SCHEMA = "event_id INT, company_id STRING, amount DOUBLE, event_ts STRING"


def run_counts(spark, source_dir, query_name):
    # TODO: readStream JSON (SCHEMA) -> groupBy("company_id").count() aliased as n ->
    # writeStream.format("memory").queryName(query_name) with the correct outputMode,
    # trigger(availableNow=True); processAllAvailable(); stop(); return the query.
    raise NotImplementedError

Hints

Hint 1

This is an aggregation (groupBy(company_id).count()). append is illegal for a running, unwatermarked aggregation because the counts keep changing — Spark will reject it.

Hint 2

Write to the in-memory sink: .writeStream.format('memory').queryName(query_name). For an aggregation that reports the whole current result, use outputMode('complete').

Hint 3

Use .trigger(availableNow=True), then processAllAvailable() and stop(). Then the caller reads SELECT * FROM <query_name>. Return the (stopped) query. The final per-company counts must match a DuckDB oracle.

Run it locally
uv run de start SPK-10/02-output-modes
# edit workspace/SPK-10/02-output-modes/
uv run de test