Spark Java处理320万行数据集：转HashMap列表遇添加失败求最优方案

阿华AIGC实验室

2026-5-11

你遇到的核心问题是Spark分布式执行模型导致的对象可见性问题：finalJsonMap是在Driver端初始化的本地ArrayList，但foreachPartition里的逻辑是在各个Executor节点上跑的——Executor没法直接修改Driver端的集合，你在Executor里往rowMap加元素的操作，只会在Executor本地内存生效，完全无法同步回Driver端的finalJsonMap，所以最后Driver拿到的还是空列表。

下面给你两种适配不同场景的最佳解决方案：

方案1：用`mapPartitions` + `collect`（适合数据量能被Driver内存承载的场景）

这是Spark里把分布式数据集转成Driver端本地集合的标准写法，会在每个Executor上处理分区内的数据，再把结果汇总回Driver：

// 指定泛型，避免原始类型警告，提升代码安全性
List<HashMap<String, String>> finalJsonMap = srcData.mapPartitions(new MapPartitionsFunction<Row, HashMap<String, String>>() {
    @Override
    public Iterator<HashMap<String, String>> call(Iterator<Row> t) throws Exception {
        // 每个分区内创建本地列表，存储当前分区的HashMap
        List<HashMap<String, String>> partitionMapList = new ArrayList<>();
        while (t.hasNext()) {
            Row eachRow = t.next();
            HashMap<String, String> rowMap = new HashMap<>();
            for (int j = 0; j < grpdColNames.size(); j++) {
                rowMap.put(grpdColNames.get(j), eachRow.getString(j));
            }
            partitionMapList.add(rowMap);
        }
        return partitionMapList.iterator();
    }
}).collect();

为什么这能解决问题？

mapPartitions是转换算子，在每个Executor的分区上独立执行，把分区内的Row转成HashMap列表；
collect()是动作算子，会把所有Executor上的结果拉取到Driver端，合并成一个完整的List，这样Driver就能拿到所有的HashMap了。

方案2：直接输出为JSON文件（大数据量场景推荐）

如果你的最终目标是生成JSON结果，完全没必要把数据拉到Driver端，Spark本身支持分布式输出JSON文件，效率更高，还能避免Driver内存溢出（320万行如果每个条目复杂，拉到Driver很容易OOM）：

// 先把Dataset的列名替换成你需要的grpdColNames（如果原列名已经匹配可以跳过这步）
Dataset<Row> renamedData = srcData.toDF(grpdColNames.toArray(new String[0]));
// 直接输出JSON，Spark会自动按分区生成文件，也可以用coalesce(1)合并成单个文件（大数据量不推荐）
renamedData.write()
           .mode(SaveMode.Overwrite)
           .json("/path/to/your/output/directory");