我在JavaPairRDD<String, MyPojo>
whereMyPojo
是一个带有属性的pojoHashSet<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
。
我在JavaPairRDD<String, MyPojo>
whereMyPojo
是一个带有属性的pojoHashSet<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
。
如果必须找到具有值交集的键,可以使用这种方法:
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)