Spark DataFrame mapPartitions

Question

Ich muss die verteilte Berechnung auf Spark DataFrame fortsetzen, wobei einige willkürliche (nicht SQL-) Logik auf Brocken von DataFrame aufgerufen wird. ich getan habe:

def some_func(df_chunk): 
    pan_df = df_chunk.toPandas() 
    #whatever logic here 

df = sqlContext.read.parquet(...) 
result = df.mapPartitions(some_func) 

Unfortunatelly es führt zu:

AttributeError: 'itertools.chain' object has no attribute 'toPandas'

ich erwartet hatte Funke Dataframe Objekt innerhalb jeder Karte Aufruf zu haben, stattdessen habe ich 'itertools.chain'. Warum? Und wie kann man das überwinden?

Answer 1

Versuchen Sie folgendes:

>>> columns = df.columns 
>>> df.rdd.mapPartitions(lambda iter: [pd.DataFrame(list(iter), columns=columns)])