Size the Executors

A medium Spark interview practice problem on DataDriven. Write the Spark code and run it against the tables, with instant feedback.

Domain
Spark
Difficulty
medium
Seniority
L5
Asked in
Microsoft

Problem

On a 50-node cluster where every machine has 128 GB and 32 cores, keeps 1 core and 1 GB for the OS, and splits its remaining memory in whole gigabytes among as many executors as its free cores allow, each executor gives up 2 GB of that share to off-heap overhead and keeps the rest as heap. The recommendations team's feature job already died on 16-core executors from GC thrash and on 4 GB executors that could not hold its 2.1 GB broadcast table, so only 1 to 5 cores per executor are in play; with no cluster table to read, number the rows of users and take positions 1 through 5 as those core counts. List the sizings whose heap is at least three times the broadcast size, with cores per executor, executors per node, heap in GB, and the executor count across the whole cluster.

Too big: GC kills you. Too small: broadcast kills you.

Expected output

First 3 of 3 rows
cores_per_executorexecutors_per_nodeheap_gbtotal_executors
31010500
4716350
5619300

Worked solution and explanation

What this really is

This is a two-sided constraint problem wearing an incident costume. The skill being probed: can you turn cores per executor into heap size by doing the node arithmetic honestly, then hold each sizing against a memory floor that pulls the opposite way from the thread cap? Anyone can say "use 5 cores". The trap is the memory you never get: the gigabyte the OS keeps and the 2 GB each executor hands to overhead. Skip them and the 2-core sizing looks like an 8 GB heap that clears the broadcast floor, when it really has 6 GB and dies exactly like Config B.

Step 1: Number the users rows to get candidate core counts

The cap of 5 threads per JVM comes straight from Config A, so the only candidates are 1 through 5. Numbering users in user_id order with ROW_NUMBER() gives a gap-free counter whatever the ids look like; keep positions up to 5 and call it cores_per_executor.

Step 2: Take the OS reserve off, then let cores decide executors

Only 31 cores and 127 GB per node are schedulable. executors_per_node is FLOOR(31 / cores_per_executor), and leftover cores sit idle: 4-core executors fit 7 per node and strand 3 cores, which is the honest price of that sizing.

Step 3: Split memory, then pay overhead

Each executor's container is FLOOR(127 / executors_per_node) whole gigabytes, and 2 GB of that is off-heap overhead, so heap_gb is the container minus 2. YARN counts the overhead against the node; your tasks just cannot use it.

Step 4: Apply the broadcast floor

Config B says the 2.1 GB broadcast needs room beside execution and storage memory, so the heap must be at least three times it: heap_gb >= 6.3. That floor cuts the 1-core and 2-core sizings and leaves three.

Enumerate and filter the sizings
BROADCAST_GB = 2.1

by_user = Window.orderBy("user_id")

candidates = (
    spark.table("users")
    .select(F.row_number().over(by_user).alias("cores_per_executor"))
    .where(F.col("cores_per_executor") <= 5)
    .withColumn("executors_per_node", F.floor(F.lit(31) / F.col("cores_per_executor")))
)

sized = candidates.withColumn(
    "heap_gb", F.floor(F.lit(127) / F.col("executors_per_node")) - 2
)

result = (
    sized
    .where(F.col("heap_gb") >= 3 * BROADCAST_GB)
    .select(
        "cores_per_executor",
        "executors_per_node",
        "heap_gb",
        (F.lit(50) * F.col("executors_per_node")).alias("total_executors"),
    )
)

Three sizings survive: 3, 4 and 5 cores, with 10, 16 and 19 GB heaps.

Naive memory split
2 cores: 16 executors per node, 128 / 16 = 8 GB heap. Clears the 6.3 GB floor, so it ships.
Honest memory split
2 cores: FLOOR(31 / 2) = 15 per node, FLOOR(127 / 15) = 8 GB container, minus 2 GB overhead = 6 GB heap. Fails the floor.

Common follow-up questions