Иван Судос Иван Судос - 4 months ago 97
Python Question

Spark DataFrame mapPartitions

I need to proceed distributed calculation on Spark DataFrame invoking some arbitrary (not SQL) logic on chunks of DataFrame.
I did:

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

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


Unfortunatelly it leads to:


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


I expected to have spark DataFrame object within each map invocation, instead I got 'itertools.chain'. Why? And how to overcome this?

Answer

Try this:

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