使用 JavaPairRDD 作为 Spark 方式实现 Hadoop Map

Question

我有一个 RDD：

JavaPairRDD<Long, ViewRecord> myRDD

通过newAPIHadoopRDD 方法创建。我有一个现有的地图功能，我想以 Spark 方式实现它：

LongWritable one = new LongWritable(1L);

protected void map(Long key, ViewRecord viewRecord, Context context)
    throws IOException ,InterruptedException {

  String url = viewRecord.getUrl();
  long day = viewRecord.getDay();

  tuple.getKey().set(url);
  tuple.getValue().set(day);

  context.write(tuple, one);
};

PS：元组来源于：

KeyValueWritable<Text, LongWritable>

可以在这里找到：TextLong.java

Answer 1

我不知道元组是什么，但是如果您只想将记录映射到键为 (url, day) 和值 1L 的元组，您可以这样做：

result = myRDD
    .values()
    .mapToPair(viewRecord -> {
        String url = viewRecord.getUrl();
        long day = viewRecord.getDay();
        return new Tuple2<>(new Tuple2<>(url, day), 1L);
    })


//java 7 style
JavaPairRDD<Pair, Long> result = myRDD
        .values()
        .mapToPair(new PairFunction<ViewRecord, Pair, Long>() {
                       @Override
                       public Tuple2<Pair, Long> call(ViewRecord record) throws Exception {
                           String url = record.getUrl();
                           Long day = record.getDay();

                           return new Tuple2<>(new Pair(url, day), 1L);
                       }
                   }
        );