Spark mapPartitions 迭代器返回重复记录答案

【问题标题】：Spark mapPartitions iterator returns duplicate recordsSpark mapPartitions 迭代器返回重复记录
【发布时间】：2018-06-13 08:50:08
【问题描述】：

我有一个 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();
    }

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

【问题讨论】：

标签： java apache-spark distributed-computing

【解决方案1】：

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

【讨论】：