如何在 Apache Spark 中重置 MapReduce 函数上的迭代器答案

【问题标题】：How to reset Iterator on MapReduce Function in Apache Spark如何在 Apache Spark 中重置 MapReduce 函数上的迭代器
【发布时间】：2023-03-18 01:31:01
【问题描述】：

我是 Apache-Spark 的新手。我想知道如何在 Apache Spark 的 MapReduce 函数中重置指向 Iterator 的指针，以便我编写

Iterator<Tuple2<String,Set<String>>> iter = arg0;

但它不起作用。以下是在 java 中实现 MapReduce 功能的类。

class CountCandidates implements Serializable,
    PairFlatMapFunction<Iterator<Tuple2<String,Set<String>>>, Set<String>, Integer>,
    Function2<Integer, Integer, Integer>{

    private List<Set<String>> currentCandidatesSet;
    public CountCandidates(final List<Set<String>> currentCandidatesSet) {
        this.currentCandidatesSet = currentCandidatesSet;
    }

    @Override
    public Iterable<Tuple2<Set<String>, Integer>> call(
            Iterator<Tuple2<String, Set<String>>> arg0)
            throws Exception {
        List<Tuple2<Set<String>,Integer>> resultList = 
                new LinkedList<Tuple2<Set<String>,Integer>>();

        for(Set<String> currCandidates : currentCandidatesSet){
            Iterator<Tuple2<String,Set<String>>> iter = arg0;
            while(iter.hasNext()){
                Set<String> events = iter.next()._2;
                if(events.containsAll(currCandidates)){
                    Tuple2<Set<String>, Integer> t = 
                            new Tuple2<Set<String>, Integer>(currCandidates,1);
                    resultList.add(t);
                }
            }
        }

        return resultList;
    }

    @Override
    public Integer call(Integer arg0, Integer arg1) throws Exception {
        return arg0+arg1;
    }
}

如果函数中无法重置迭代器，我该如何迭代参数 arg0 多次？我已经尝试了一些不同的方法作为以下代码，但它也不起作用。以下代码似乎“resultList”的数据比我预期的要多。

        while(arg0.hasNext()){
            Set<String> events = arg0.next()._2;
            for(Set<String> currentCandidates : currentCandidatesSet){
                if(events.containsAll(currentCandidates)){
                    Tuple2<Set<String>, Integer> t = 
                            new Tuple2<Set<String>, Integer>(currentCandidates,1);
                    resultList.add(t);
                }
            }
        }

我该如何解决？

提前感谢您的回答，并为我糟糕的英语感到抱歉。如果您不明白我的问题，请发表评论

【问题讨论】：

标签： java hadoop mapreduce apache-spark hadoop-yarn

【解决方案1】：

Iterator 甚至不能在普通的 Java 或 Scala 中“重置”。这就是Iterator 的本质。 Iterable 可以多次为您提供Iterators。您的代码需要重写以接受Iterable，如果这是您真正想要做的。

【讨论】：

我知道 'Iterator' 不可重置，但 mapreduce 函数强制我获取 'Iterator' 对象，所以我想知道如何解决这个问题。
您可以随时将Iterator 的内容读入Collection，但这当然意味着您必须一次将所有内容保存在内存中。这对于您的用例可能合适，也可能不合适。
你确定不能通过复制Iterable.iterator将迭代器重置到开头吗？我依稀记得这样做
猜猜不是hadoop-475
你是对的，@aaronman。该链接对我理解他们为什么不提供可克隆非常有帮助。我认为最好选择你的答案。

【解决方案2】：

理论上，如果它是可克隆的，hadoop 迭代器可以重置到开头。在 Mapreduce 框架中重置到开头是可以接受的，因为您仍然可以从头开始读取文件，从而获得更好的整体速度。将迭代器重置为随机点将与 Mapreduce 思维模式背道而驰，因为它可能需要从文件进行随机访问。

Hadoop's Jira 中有一张票，解释了为什么他们选择不使迭代器可克隆，尽管它确实表明它有可能是因为值将不必必须存储在记忆。

【讨论】：