Verwenden von Pythons reduce() zum Verbinden mehrerer PySpark DataFrames

Question

Weiß jemand, warum die Verwendung von functools.reduce() von Python3 zu einer schlechteren Performance beim Verbinden mehrerer PySpark DataFrames führen würde als nur iteratives Verbinden derselben Datenrahmen mit einer for-Schleife? Insbesondere ergibt dies eine massive Verlangsamung durch eine Out-of-Memory-Fehler folgt:

def join_dataframes(list_of_join_columns, left_df, right_df): 
    return left_df.join(right_df, on=list_of_join_columns) 

joined_df = functools.reduce(
    functools.partial(join_dataframes, list_of_join_columns), list_of_dataframes, 
) 

während dieses nicht:

joined_df = list_of_dataframes[0] 
joined_df.cache() 
for right_df in list_of_dataframes[1:]: 
    joined_df = joined_df.join(right_df, on=list_of_join_columns) 

Irgendwelche Ideen wäre sehr dankbar. Vielen Dank!

Answer 1

Ein Grund ist, dass eine Reduzierung oder eine Faltung normalerweise funktional rein ist: Das Ergebnis jeder Akkumulationsoperation wird nicht in denselben Teil des Speichers geschrieben, sondern in einen neuen Speicherblock.

Im Prinzip könnte der Garbage Collector den vorherigen Block nach jeder Akkumulation freigeben, aber wenn dies nicht der Fall ist, reservieren Sie Speicher für jede aktualisierte Version des Akkumulators.