pysparkでmapPartitionsを使用する方法

Question

Apache Sparkのドキュメントに続いて、mapPartitionモジュールを試してみました。次のコードでは、関数内で最初のRDDを見ることが期待されました。myfunc値を出力した後にイテレータを返すだけです。しかし、RDDでcollectを実行すると、それは空です。

from pyspark import SparkConf 
from pyspark import SparkContext               

def myfunc(it): 
    print(it.next()) 
    return it 

def fun1(sc): 
    n = 5 
    rdd = sc.parallelize([x for x in range(n+1)], n) 
    print(rdd.mapPartitions(myfunc).collect()) 


if __name__ == "__main__":                 
    conf = SparkConf().setMaster("local[*]")            
    conf = conf.setAppName("TEST2")              
    sc = SparkContext(conf = conf)               
    fun1(sc) 

Answer 1

mapPartitionsはここでは関係ありません。イテレータ（ここではitertools.chain）はステートフルなので、1回だけトラバースできます。 it.next()に電話すると、最初の要素を読み込んで破棄し、返されるのはシーケンスの末尾です。

パーティションにアイテムが1つしかない場合（1つを除いてすべてのパーティションが対象です）、パーティション全体を効果的に破棄します。

いくつかの注意事項：

• は、タスク内stdoutに何を置く一般的に無用です。
• nextの使い方は移植性が低く、Python 3では使用できません。