java - Spark mapPartitions 迭代器返回重复记录

Question

我有一个 FlatMapFunction>> 的实现类，String>。为每个分区初始化一些不可串行的连接。但是当我在迭代器上调用 next() 时，它为多个分区提供相同的记录。代码如下:

@Override
    public Iterator < String > call(Iterator < Tuple2 < String, Iterable < String >>> tuple2Iterator)
    throws Exception {
        BitLambdaService lambda = buildClient();
        List <String> resultList = new ArrayList < > ();
        while (tuple2Iterator.hasNext()) {
            Tuple2 < String, Iterable < String >> tpl = tuple2Iterator.next();
            // do something
        }
        return resultList.iterator();
    }

有人之前遇到过这个问题吗？或者知道如何解决它？

Answer 1

通过调用rdd.cache()解决了这个问题就在转型之后。出现该问题的原因是转换是以惰性方式执行的，并且当将操作应用于 RDD 时，转换实际上是在将操作应用于 RDD 时执行的。因此，mapPartitions 不会等待调用方法完成，而是将相同的记录分配给另一个执行程序。