Как использовать фильтр с помощью containsAll и содержит в javapairrdd

Question

У меня есть 2 коллекции, одна из которых "список", а другая "pairRdd2", которая содержит данные, как указано ниже.

Я пытаюсь применить фильтр с containsAll, где, если mypairRdd2 содержит все значения, указанные в списке. Ожидаемый результат - Джо,{США, Великобритания}

List<String> list = Arrays.asList("US","UK");

JavaRDD pairRdd = ctx.parallelize(Arrays.asList(new Tuple2("john","US"),new Tuple2("john","UAE"),new Tuple2("joe","US"),new Tuple2("joe","UK")));

JavaPairRDD<String, String> pairRdd2 = JavaPairRDD.fromJavaRDD(pairRdd);

pairRdd2.groupByKey().filter(x-> Arrays.asList(x._2).containsAll(list)).foreach(new VoidFunction<Tuple2<String,Iterable<String>>>() {

    @Override
    public void call(Tuple2<String, Iterable<String>> t) throws Exception {
        System.out.println(t._1());             
    }
});

Может кто-то выделить, что я делаю не так...

Answer 1

Проблема с Arrays.asList(), Это создает список Iterableс, что не то, что вам нужно для выполнения фильтра. Вы должны использовать список, данный groupBy сам:

    pairRdd2.groupByKey().filter(f -> {
        Set<String> set = new HashSet<>();
        for(String s: f._2())
            set.add(s);

        return list.containsAll(set);
    });

Вы также можете найти быстрый способ конвертировать итерируемый / итератор в коллекцию и вообще избежать цикла.