Javaを使用したRDD関数の型の不一致

Question

私はスパークするのが初めてで、簡単なコードを作成しましたが、タスクを実行中に問題が発生しました。私はJava 1.7 SEとSpark 2.0.1を使っていくつかのタスクを実行しています。

以下は私のコードです。

JavaRDD<String> newLine = line.flatMap(new FlatMapFunction<String, String>() { 
    public List<String> call(String s) throws Exception { 
     List<String> ret = new ArrayList<String>(); 
     List<String> ls = Arrays.asList(s.split("\t")); 

     String values = ls.get(ls.size()-1); 
     List<String> value = Arrays.asList(values.split("\\|")); 

     for(int i=0;i<value.size();++i){ 
      String ns = ls.get(0)+"\t"+ls.get(1)+"\t"+ls.get(2)+"\t"+ls.get(3)+"\t"+ls.get(4)+"\t"+ls.get(5); 
      ns = ns + "\t" + value.get(i); 
      ret.add(ns); 
     } 
     return ret; 
    } 
}); 

エラーは、List<String>のpublic List<String> call(String s) throws Exceptionで起こりました。エラーメッセージは、互換性のない戻り値の型を使用しようとしたことを示しています。私はこのメッセージを理解しましたが、適切な解決策を見つけることができませんでした。さらに、このエラーはmavenリポジトリを使用してsparkライブラリ2.0.1を1.6.0から変更したときに発生しました。 この問題の解決方法に関するガイダンスを私に教えていただければ幸いです。

以下のように便宜のためにコード全体を添付しました。

import java.util.ArrayList; 
import java.util.Arrays; 
import java.util.Iterator; 
import java.util.List; 
import org.apache.spark.SparkConf; 
import org.apache.spark.api.java.JavaRDD; 
import org.apache.spark.api.java.JavaSparkContext; 
import org.apache.spark.api.java.function.FlatMapFunction; 

public class DataTrans { 
    public static void main(String[] args) throws Exception{ 
     String inputFile = args[0]; 
     String outputFile = args[1]; 

     SparkConf conf = new SparkConf().setAppName("Data Transformation") 
       .set("spark.serializer","org.apache.spark.serializer.KryoSerializer").setMaster("local[*]"); 

     JavaSparkContext sc = new JavaSparkContext(conf); 

     /* Load our input data. */ 
     JavaRDD<String> input = sc.textFile(inputFile); 

     JavaRDD<String> line = input.flatMap(new FlatMapFunction<String, String>() { 
      public Iterator<String> call(String s) throws Exception { 
       return Arrays.asList(s.split("\n")).iterator(); 
      } 
     }); 

     JavaRDD<String> newLine = line.flatMap(new FlatMapFunction<String, String>() { 
      public List<String> call(String s) throws Exception { 
       List<String> ret = new ArrayList<String>(); 
       List<String> ls = Arrays.asList(s.split("\t")); 

       String values = ls.get(ls.size()-1); 
       List<String> value = Arrays.asList(values.split("\\|")); 

       for(int i=0;i<value.size();++i){ 
        String ns = ls.get(0)+"\t"+ls.get(1)+"\t"+ls.get(2)+"\t"+ls.get(3)+"\t"+ls.get(4)+"\t"+ls.get(5); 
        ns = ns + "\t" + value.get(i); 
        ret.add(ns); 
       } 
       return ret; 
      } 
     }); 
     newLine.saveAsTextFile(outputFile); 

     sc.stop(); 
     sc.close(); 
    } 
} 

おかげ

Answer 1

それはIterator<String>ないList<String>呼び出しで（）の戻り値の型

JavaRDD<String> newLine = line.flatMap(new FlatMapFunction<String, String>() { 
     public Iterator<String> call(String s) throws Exception { 
      List<String> ret = new ArrayList<String>(); 
      List<String> ls = Arrays.asList(s.split("\t")); 

      String values = ls.get(ls.size()-1); 
      List<String> value = Arrays.asList(values.split("\\|")); 

      for(int i=0;i<value.size();++i){ 
       String ns = ls.get(0)+"\t"+ls.get(1)+"\t"+ls.get(2)+"\t"+ls.get(3)+"\t"+ls.get(4)+"\t"+ls.get(5); 
       ns = ns + "\t" + value.get(i); 
       ret.add(ns); 
      } 
      return ret.iterator(); 
     } 
    });