Apache Spark mapPartitionsWithIndex

Question

有人可以给出在Java中正确使用mapPartitionsWithIndex的示例吗？我发现了很多Scala示例，但缺乏Java示例。 我的理解是正确的，使用此函数时，单独的分区将由单独的节点处理。

我收到以下错误

method mapPartitionsWithIndex in class JavaRDD<T> cannot be applied to given types; 
    JavaRDD<String> rdd = sc.textFile(filename).mapPartitionsWithIndex 
    required: Function2<Integer,Iterator<String>,Iterator<R>>,boolean 
    found: <anonymous Function2<Integer,Iterator<String>,Iterator<JavaRDD<String>>>> 

在做

JavaRDD<String> rdd = sc.textFile(filename).mapPartitionsWithIndex(
    new Function2<Integer, Iterator<String>, Iterator<JavaRDD<String>> >() { 

    @Override 
    public Iterator<JavaRDD<String>> call(Integer ind, String s) { 

Answer 1

下面是我用它来删除一个CSV文件的第一行代码：

JavaRDD<String> rawInputRdd = sparkContext.textFile(dataFile); 

Function2 removeHeader= new Function2<Integer, Iterator<String>, Iterator<String>>(){ 
    @Override 
    public Iterator<String> call(Integer ind, Iterator<String> iterator) throws Exception { 
     if(ind==0 && iterator.hasNext()){ 
      iterator.next(); 
      return iterator; 
     }else 
      return iterator; 
    } 
}; 
JavaRDD<String> inputRdd = rawInputRdd.mapPartitionsWithIndex(removeHeader, false);