使用mapPartitionsToPair/PairFlatMapFunction时如何返回迭代器

Question

在spark中使用mapPartitionsToPair/PairFlatMapFunction时，我在Internet上找到了一个类似的例子

spark.read ().textFile (hdfsPath).javaRDD ()
.mapPartitionsToPair (new PairFlatMapFunction <Iterator <String>, String, String> () {
  public Iterable <Tuple2 <String, String> > call (Iterator <String> input) {
    List <String> result = new ArrayList <String> ();
    while (input.hasNext ()) result.add (doSomeThing (input.next ()));
    return result;
  }
});

但当康普利

return type Iterable<Tuple2<String,String>> is not compatible with Iterator<Tuple2<String,String>>

我找到了call的声明

java.util.Iterator<scala.Tuple2<K,V>> call(T t) 

所以调用应该是返回一个迭代器。

因此，有人能帮助我如何返回在javaRDD api火花迭代器？谢谢

PS：我试过下面这样的代码，但在集群上不起作用：

public Iterator <Tuple2 <String, Strng> > call (Iterator <String> input) {
  List <String> result = new ArrayList <String> ();
  while (input.hasNext ()) result.add (doSomeThing (input.next ()));
  return result.iterator;
}

Answer 1

看起来您的开发环境和集群之间的spark版本不匹配。

从Spark-2.0.0，JavaRDD的平面图和映射分区函数返回Java迭代器，不可迭代。

因此，如果你的集群小于Spark-2.0.0，那么在开发时也使用相同的Spark版本。

对于Spark-2.0.0或更高版本，

public Iterator <Tuple2 <String, Strng> > call (Iterator <String> input) {
List <String> result = new ArrayList <String> ();
while (input.hasNext ()) result.add (doSomeThing (input.next ()));
return result.iterator;
}

应该行得通。

对于小于2.0.0的spark版本，

public Iterable <Tuple2 <String, String> > call (Iterator <String> input) {
List <String> result = new ArrayList <String> ();
while (input.hasNext ()) result.add (doSomeThing (input.next ()));
return result;
}

应该行得通。