スパーク行にあるキーでJavaPairRDDにcsvをロード

Question

キーとして行の値を使用し、値として行自体を使用して、JavaPairRDDにcsvをロードします。現在、私はこのようにそれをやっている：

public class FactData implements Serializable{ 
    public String key; 
    public int m1; 
    public int m2; 
    public int m3; 
} 

私は」：私はこのようなこれらの行を表すJavaオブジェクトを持っている

a,1,1,2 
b,1,1,2 
a,2,2,3 
b,2,2,3 

：

私はこのようなラインを持っているcsvファイルを持っています現在、このようなペアRDDになっているm：

JavaRDD<FactData> lines = sc.textFile("test.csv").map(line -> FactData.fromFileLine(line)); 

JavaPairRDD<String, Iterable<FactData>> groupBy = lines.groupBy(row -> row.getId()); 

しかし、もっと速く/より良い方法があるかどうか疑問に思っています この？次のようなもの：

JavaPairRDD<String,Iterable<FactData>> groupedLines = sc.textFile("test.csv").flatMapToPair(new PairFlatMapFunction<String, String, Iterable<FactData>>() { 
    @Override 
    public Iterator<Tuple2<String, Iterable<FactData>>> call(String s) throws Exception { 
     //WHAT GOES IN HERE? 
     return null; 
    } 
}); 

Answer 1

なぜkeyByを使用しないのですか？

たとえば、ファイルの最初の値をキーとして、合計行を値として設定します。 これを簡単に行うことができます。

JavaRDD<String> lines = context.textFile("test.csv"); 
JavaPairRDD<String, String> newLines = lines.keyBy(new Function<String,String>(){ 
     @Override 
     public String call(String arg0) throws Exception { 
     return arg0.split(",")[0]; 
     } 
    }); 

マップとして収集したい場合は、これを行うことができます。

JavaPairRDD<String, Iterable<String>> newLines = lines.keyBy(new Function<String,String>(){ 

      @Override 
      public String call(String arg0) throws Exception { 
       return arg0.split(",")[0]; 
      }   
     }).mapValues(new Function<String, Iterable<String>>(){ 
      @Override 
      public Iterable<String> call(String arg0) throws Exception { 
       return Arrays.asList(arg0.split(",")); 
      } 
     });