Совокупные значения Iterable с Spark API

Question

Я экспериментирую Spark с Java 8, чтобы провести анализ CSV-файла.

Я хочу вычислить сумму элементов по ключевым словам.

My RDD имеет тип JavaPairRDD<String, Iterable<Record>>, и я хочу преобразовать это RDD в JavaPairRDD<String, Integer>, где целочисленное значение является суммой всех элементов в итерируемой коллекции.

Запись - это простой java POJO.

Любая идея о том, как это сделать?

JavaPairRDD<String, Iterable<Record>> distinct = map 
    .filter(record -> record.circonference > 170) 
    .groupBy(record -> record.espece) 
    .distinct(); 

Мой POJO класс

public static class Record implements Serializable { 

    private String geoPoint; 
    private float circonference; 
    private String addresse; 
    private float hauteur; 
    private String espece; 
    private String variete; 
    private String datePlantation; 

Answer 1

Предполагая, что ваш вход РДД называется рдд,

rdd.mapValues(records -> { 
    Integer ret = 0; 
    for(Record r: records) { 
     ret += Math.round(ret.circonference); 
     ret += Math.round(ret.hauteur); 
    } 
    return ret; 
}) 

Я не совсем уверен, почему вы хотели бы суммировать все поплавки в целое число, но это должно более или менее делать трюк для вас. mapValues позволяет нам применить функцию к части значений пары ключ/значение, сохраняя один и тот же ключ, чтобы он соответствовал этой проблеме.

Answer 2

Решено мной с помощью преобразования mapToPair.

Мы можем сопоставить второй элемент с размером итерации коллекции, как это:

JavaPairRDD<String, Iterable<Record>> distinct = map 
    .filter(record -> record.circonference > 170) 
    .groupBy(record -> record.espece) 
    .mapToPair(t -> new Tuple2<String,Integer>(t._1, Lists.newArrayList(t._2).size()))