DE Mastery GitHub

Apache Spark · SPK-06

Joins and aggregations at scale

From your world. Redshift DISTSTYLE ALL ≈ broadcast; distribution keys and skew

← All exercises

Broadcast variable and accumulator

Spark gives you two low-level primitives for shared state. They are separate from the broadcast join:

  • A broadcast variable sends a read-only value to every executor once.
  • An accumulator is a counter for metrics. Executors write to it, and the driver can read it.

Here you use both. You validate the orders feed against an allowlist of valid currencies. At the same time, you count how many rows had a bad code.

Task

Implement validate_currencies(spark, orders, valid_currencies). It must do these steps:

  1. Broadcast valid_currencies (a Python set/list) to the executors.
  2. Iterate over the rows (with orders.rdd). For each row whose currency is not in the broadcast allowlist, increment an accumulator.
  3. Return a tuple (valid_df, invalid_count). valid_df contains exactly the rows whose currency is in the allowlist (same schema as orders). invalid_count is the accumulator value.

Constraints

  • Read the allowlist through the .value of the broadcast variable. Do not use the original Python object that you captured directly.
  • Trigger an action, so that the accumulator has its values before you read it. Accumulators are reliable only after an action. Treat them as monitoring, not business logic.

How it is graded

  • Data (DuckDB oracle): valid_df equals SELECT * FROM orders WHERE currency IN (...).
  • Counter: invalid_count equals the number of orders whose currency is not in the allowlist (DuckDB count). That is, valid + invalid == total.

Starter code

solution.py
"""SPK-06 / 03-broadcast-var-accumulator — starter. See README.md."""


def validate_currencies(spark, orders, valid_currencies):
    # TODO: broadcast valid_currencies; iterate orders.rdd; increment an
    # accumulator for rows whose currency is NOT in the allowlist; trigger an
    # action; return (valid_df, invalid_count).
    raise NotImplementedError

Hints

Hint 1

Use spark.sparkContext.broadcast(valid_currencies) to ship the allowlist to executors, and spark.sparkContext.accumulator(0) to count bad-currency rows from inside the RDD map.

Hint 2

Work on orders.rdd: for each row, read the broadcast value (.value) to decide valid vs invalid; add 1 to the accumulator for invalid rows. Return (valid_df, invalid_count).

Hint 3

Return a tuple: a DataFrame of the valid rows (same columns as orders) and the integer accumulator value. Trigger an action (e.g. build the valid DF via collect/count path) so the accumulator is populated before you read it.

Run it locally
uv run de start SPK-06/03-broadcast-var-accumulator
# edit workspace/SPK-06/03-broadcast-var-accumulator/
uv run de test