Spark DataFrame mapPartitions

Question

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 1

Try this:

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