PySpark apply function on 2 dataframes and write to csv for billions of rows on small hardware

There are a couple of things you can do to improve your computation:

Improve parallelism

As Nithish mentioned in the comments, you don't have enough partitions in your input data frames to make use of all your CPU cores. You're not using all your CPU capability and this will slow you down.

To increase your parallelism, repartition dfc to at least your number of cores:

dfc = dfc.repartition(dfc.sql_ctx.sparkContext.defaultParallelism)

You need to do this because your crossJoin is run as a BroadcastNestedLoopJoin which doesn't reshuffle your large input dataframe.

Separate your computation stages

A Spark dataframe/RDD is conceptually just a directed action graph (DAG) of operations to run on your input data but it does not hold data. One consequence of this behavior is that, by default, you'll rerun your computations as many times as you reuse your dataframe.

In your fuzzy_match_approve function, you run 2 separate filters on your df, this means you rerun the whole cross-join operations twice. You really don't want this !

One easy way to avoid this is to use cache() on your fuzzy_match result which should be fairly small given your inputs and matching criteria.

def fuzzy_match_running(dfs, dfc, path_summary):
    """
    Implements the Levenshtein and Soundex algorithms and returns a fuzzy matched DataFrame.
    Filters out those where resulting LS distance is less than 15% of SF name length.
    """
    # Apply Levenshtein and Soundex functions
    dfs = dfs.withColumn("OrganisationNameKeyLen", F.length("OrganisationNameKey")).cache()
    dfc = dfc.repartition(dfc.sql_ctx.sparkContext.defaultParallelism).cache()
    
    df = dfc.crossJoin(dfs) \
        .withColumn( "LevenshteinDistance", F.levenshtein( F.lower("OrganisationNameKey") , F.lower("CompanyNameKey") ) ) \
        .withColumn( "HasSameSoundex", F.soundex("OrganisationNameKey") == F.soundex("CompanyNameKey") ) \
        .where("LevenshteinDistance < OrganisationNameKeyLen * 0.15") \
        .orderBy("OrganisationName", "CompanyName") \
        .cache()
    return df

If I run my fuzzy_match_running on some example data frames on my 8 core/16 threads I9-9980HK laptop (spark in local[*] mode with 8GB driver memory):

dfc rowcount : 572494
dfs rowcount : 17728
fuzzy_match rowcount: 7228499
Duration: 679.5572581291199 seconds
Matches/core/sec: 933436.210726889

The job takes about 12 min doing 572494*17728 ~ 10 billion row comparisons at 933k comparisons/seconds/core. Since your job does 64 billions row comparisons I would expect it to take about 80 min on my laptop.

You should run a similar experiment on your computer with a smaller sample to get an idea of your actual computing speed.

Going further: maximizing matches/sec

To go faster, we need to adjust the computation and increase the number of comparisons that can be done per seconds. A few things stand out in the function:

• you filter your output by comparing the levenshtein distance, an integer, to a decimal calculation. This means spark will cast your integer to a decimal and operate on decimal. Comparing decimals is much slower than integers and it's unnecessary here, you can cast the bound to an int beforehand.
• your levenshtein operates on the lower versions of your keys, this means, for each row comparison, Spark will convert the column values to lower again and again, wasting CPU cycles for redundant stuff. You can preprocess this before your join.

I update the function like this:

def fuzzy_match(dfs: DataFrame, dfc: DataFrame, path_summary: str) -> DataFrame:
    dfs = dfs.withColumn("OrganisationNameKeyLower", F.lower("OrganisationNameKey"))\
        .withColumn("MatchingTolerance", F.ceil(F.length("OrganisationNameKey") * 0.15).cast("int"))\
        .cache()

    dfc = dfc.repartition(dfc.sql_ctx.sparkContext.defaultParallelism)\
        .withColumn("CompanyNameKeyLower", F.lower("CompanyNameKey"))\
        .cache()

    df = dfc.crossJoin(dfs)\
        .withColumn("LevenshteinDistance", F.levenshtein(F.col("OrganisationNameKeyLower"), F.col("CompanyNameKeyLower")).cast("int")) \
        .where("LevenshteinDistance < MatchingTolerance")\
        .drop("MatchingTolerance")\
        .cache()

    # clean unnecessary caches before returning
    dfs.unpersist()
    dfc.unpersist()
    return df

When running the updated version on the same inputs as before and on the same computer I get nearly twice the performance as the first implementation

dfc rowcount : 572494
dfs rowcount : 17728
fuzzy_match rowcount: 7228499
Duration: 356.23311281204224 seconds
Matches/core/sec: 1780641.1846241967

If that is still too slow for your needs, you'll need to find conditions on your data that you can use as a join condition but that's highly data and use case specific.