java - 在 Java Spark 中迭代大型数据集的最快有效方法

Question

我正在使用以下方法将 spark 数据集转换为 HashMap 列表，我的最终目标是构建 json 对象列表或 HashMap 列表我在 320 万行上运行此代码

List<HashMap> finalJsonMap = new ArrayList<HashMap>();
    srcData.foreachPartition(new ForeachPartitionFunction<Row>() {
        public void call(Iterator<Row> t) throws Exception {
            while (t.hasNext()){
                Row eachRow = t.next();
                HashMap rowMap = new HashMap();
                for(int j = 0; j < grpdColNames.size(); j++) {
                    rowMap.put(grpdColNames.get(j), eachRow.getString(j));  
                }
                finalJsonMap.add(rowMap);
            }
        }
    });

迭代工作正常，但我无法将 rowMap 添加到 finalJsonMap 中。

执行此操作的最佳方法是什么？

Answer 1

这确实不是 Spark 的工作方式。

foreachPartition 中的代码在与原始代码不同的上下文中执行

List<HashMap> finalJsonMap = new ArrayList<HashMap>();

您在此类设置中所能做的就是修改本地副本。

这已在 Stack Overflow 上多次讨论，并在 official documentation in the Understanding Closures 中有详细描述。部分。

考虑到所需的结果(即本地集合)，除了将代码转换为使用 mapPartitions 和 collect 之外，您别无他法。然而，这在 Spark 中几乎没有效率或惯用。

我强烈建议重新考虑您当前的设计。