Spark DataFrame mapPartitions

Question

我需要在Spark DataFrame上进行分布式计算,在DataFrame的块上调用一些任意(不是SQL)逻辑.我做了:

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

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

不幸的是,它导致:

AttributeError:'itertools.chain'对象没有属性'toPandas'

我希望在每个地图调用中都有spark DataFrame对象,而不是'itertools.chain'.为什么？以及如何克服这一点？

Answer 1

试试这个:

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