Каково правильное преобразование JavaRDD в кластерные строки на непересекающихся множествах?

Question

У меня есть настройки строк в JavaPairRDD<String, MyPojo>, где MyPojo - это pojo с атрибутом (назовем его HashSet<String> values).

Теперь я хочу сгруппировать (объединить) мои строки на основе любого пересечения с MyPojo.values.

Например:

<Row K1 : MyPojo (values: [A,B,C])>

<Row K2 : MyPojo (values: [A,B])>

<Row K3 : MyPojo (values: [D,E,F])>

Я хочу объединить строки с ключами K1, K2.

Answer 1

Если необходимо найти ключи с пересечением значений, можно использовать такой подход:

    List<Tuple2<String, MyPojo>> data = Lists.newArrayList(
            new Tuple2("K1", new MyPojo("A", "B", "C")),
            new Tuple2("K2", new MyPojo("A", "B")),
            new Tuple2("K3", new MyPojo("D", "E", "F")));
    JavaPairRDD<String, MyPojo> original = jsc().parallelizePairs(data);

    JavaPairRDD<String, String> preparedToJoin = original.flatMapToPair(
            v ->
                    v._2().getValues().stream().map(
                            s -> new Tuple2<String, String>(s, v._1()))
                            .collect(Collectors.toList()).iterator()
    );

    preparedToJoin.join(preparedToJoin)
            .filter(v -> !v._2()._1().equals(v._2()._2()))
             // remove one of: (K1,K2), (K2,K1)
            .filter(v -> v._2()._1().compareTo(v._2()._2()) <= 0)
            .values()
            .distinct().foreach(v -> System.out.println(v));

Вывод:

(K1,K2)