[Python] Optimize PySpark code on large dataframe that exceeds cluster resources

I have a large PySpark dataframe containing 250 million rows, with only 2 columns. I am running the minHash code found here. I tried to write the resulting dataframe to parquet files by adj_sdf.write.mode("append").parquet("/output/folder/"). However, I kept getting the error pod ephemeral local storage usage exceeds the total limit of containers. I can't increase the cluster resources, so I'm wondering if there are ways to optimize the PySpark code instead.

So far, I've done the following:

1. Partition the dataframe before running the minHash function: sdf = sdf.repartition(200)
2. Filter out pairs that are unlikely to share many hash values before the final step that involves two joins (hash_sdf.alias('a').join(...)): filtered_sdf = hash_sdf.filter(f.size(f.col('nodeSet')) > threshold), where threshold = int(0.2 * n_draws)
3. Set the number of shuffle partitions: spark.conf.set("spark.sql.shuffle.partitions", "200")

What else can I do to write the dataframe to parquet files without running into resource problems?

Continue reading...