A Job That Ran Fine Across the Cluster Crashes on the Last Line

Your pipeline processes 800GB of data spread across a 20-executor cluster. The Spark UI shows every stage completing successfully — filters, joins, a large groupBy aggregation all finish within their expected time. Then the job crashes with:

java.lang.OutOfMemoryError: Java heap space

The stack trace points to this final line:

final_df = aggregated.orderBy(F.desc("total_revenue"))
results = final_df.collect()   # <-- crash happens here
for row in results:
    send_to_dashboard_api(row)

1. Explain why a job that successfully processed 800GB across 20 executors can crash on this specific line, and why the crash is attributed to "Java heap space" rather than an executor running out of resources.
2. aggregated was the output of a groupBy that reduced 800GB down to some number of rows. Does the row count of aggregated matter to this diagnosis? Why or why not, precisely?
3. Propose two different fixes appropriate for two different underlying scenarios: one where aggregated genuinely has far too many rows to reasonably act on one at a time in a Python loop, and one where it doesn't.