Map Partition Iterator return

Question

mapPartitionsに返されるIterator listWords（）メソッドを受け入れる手助けがあります。

object MapPartitionExample { 

    def main(args: Array[String]): Unit = { 

    val conf= new SparkConf().setAppName("MapPartitionExample").setMaster("local[*]") 
    val sc= new SparkContext(conf) 

    val input:RDD[String] = sc.parallelize(List("ABC","DEF","GHU","YHG")) 

    val x= input.mapPartitions(word => listWords(word)) 


    } 

    def listWords(words: Iterator[String]) : util.Iterator[String] = { 

    val arrList = new util.ArrayList[String]() 
    while(words.hasNext) { 
     arrList.add(words.next()) 
    } 
    return arrList.iterator() 
    } 

} 

Answer 1

Iterable[NotInferU]が期待されていますが、それがあるようです残りのコードの

def listWords(words: Iterator[String]) : Iterator[String] = { 
    val arrList = new util.ArrayList[String]() 
    while(words.hasNext) { 
     arrList.add(words.next()) 
    } 
    import scala.collection.JavaConversions._ 
    return arrList.toList.iterator 
    } 

以下のようscala.collection.JavaConversions._をインポートすることにより

あなたはscala Iteratorにjava.util.Iteratorを変換する必要がありますjava.util.Iterator[String]を戻ってきています。

私は答えがmapPartitionsで使用される機能の有用

Answer 2

戻り値の型がscala.collection.Iterator、ないjava.util.Iteratorであるべきであると思います。私はあなたの現在のコードの多くのポイントが表示されていないが、あなたはScalaの可変コレクションを使用することができます。

import scala.collection.mutable.ArrayBuffer 

def listWords(words: Iterator[String]) : Iterator[String] = { 
    val arr = ArrayBuffer[String]() 
    while(words.hasNext) { 
    arr += words.next() 
    } 
    arr.toIterator 
} 

個人的に私は思いますちょうどmap：

def listWords(words: Iterator[String]) : Iterator[String] = { 
    // Some init code 
    words.map(someFunction) 
}