Spark Executor创建Dataset触发NPE，求按行大小拆分分区方案

我需要按行大小（每行含Blob，大小不一）拆分RDD，通过foreachPartition在分区内按大小分组，每组转成Dataset后写入存储，但执行sparkSession.read().json(dataset)时触发空指针异常（NPE）。已知不能在Executor中创建Dataset，求无需shuffle或repartition的按大小拆分分区的方案。

调用代码：

JavaRDD<String> stringRDD = xmlDataSet.toJavaRDD().map(xmlRow -> extractXMLBlobToJsonString(xmlRow)).rdd().toJavaRDD();
GroupBySizeManager mgr = GroupBySizeManager.getInstance();
jsonRDD.foreachPartition(mgr::splitPartition);

GroupBySizeManager.java的splitPartition方法：

public void splitPartition(Iterator<String> rowsInPartition) {
    System.out.println("********Number of groups " + this.getGroups().size());
    SparkSession sparkSession = SparkDataflowConfig.getInstance().getSparkSession();
    while (rowsInPartition.hasNext()) {
        addToGroup(rowsInPartition.next());         
    }
    
    for(List<String> rows: getRows()) {
        System.out.println("******Size of List: " + rows.size());
        Dataset<String> dataset = sparkSession.createDataset(rows, Encoders.STRING());
        Dataset<Row> xmldf = sparkSession.read().json(dataset);// NPE here
         System.out.println("Count of xmlfiltered: " + xmldf.count());
         xmldf.write().format("csv").option("header", "true").save("/home/myfilepath/");             
    }
}

异常栈：

java.lang.NullPointerException
at org.apache.spark.sql.execution.SparkPlan.sparkContext(SparkPlan.scala:67)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176)
at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:132)
at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:131)
at org.apache.spark.sql.execution.datasources.json.TextInputJsonDataSource$.inferFromDataset(JsonDataSource.scala:104)
at org.apache.spark.sql.DataFrameReader.$anonfun$json$1(DataFrameReader.scala:567)
at scala.Option.getOrElse(Option.scala:189)
at org.apache.spark.sql.DataFrameReader.json(DataFrameReader.scala:567)
at com.bm.xchange.spark.executor.GroupBySizeManager.splitPartition(GroupBySizeManager.java:57)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$foreachPartition$1(JavaRDDLike.scala:219)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$foreachPartition$1$adapted(JavaRDDLike.scala:219)
at org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2(RDD.scala:1020)

问题根源 #

SparkSession是Driver端专属的核心对象，序列化后传到Executor的实例没有绑定Executor的SparkContext，无法执行分布式计算相关操作（比如创建Dataset、执行read.json），这就是NPE的直接原因——SparkPlan找不到有效的sparkContext实例。

解决方案 #

核心思路：在Executor的分区处理逻辑中完全使用RDD/本地IO API，避免创建Dataset。以下是两种可行方案：

方案1：分区内分组后直接写文件（推荐） #

修改splitPartition方法，去掉Dataset相关逻辑，用HDFS API直接将分组后的JSON转为CSV写入文件，文件名加入分区ID和分组序号避免冲突：

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public void splitPartition(Iterator<String> rowsInPartition) {
    int partitionId = org.apache.spark.TaskContext.get().partitionId();
    ObjectMapper objectMapper = new ObjectMapper();
    FileSystem fs = FileSystem.get(org.apache.spark.SparkEnv.get().hadoopConfiguration());
    
    List<List<String>> groups = new ArrayList<>();
    long currentGroupSize = 0;
    long maxGroupSize = 1024 * 1024 * 10; // 自定义每组最大大小，示例为10MB
    List<String> currentGroup = new ArrayList<>();
    
    // 按大小分组逻辑
    while (rowsInPartition.hasNext()) {
        String row = rowsInPartition.next();
        long rowSize = row.getBytes().length;
        if (currentGroupSize + rowSize > maxGroupSize && !currentGroup.isEmpty()) {
            groups.add(currentGroup);
            currentGroup = new ArrayList<>();
            currentGroupSize = 0;
        }
        currentGroup.add(row);
        currentGroupSize += rowSize;
    }
    if (!currentGroup.isEmpty()) {
        groups.add(currentGroup);
    }
    
    // 遍历分组写入CSV
    int groupIdx = 0;
    for (List<String> group : groups) {
        String filePath = "/home/myfilepath/part-" + partitionId + "-group-" + groupIdx + ".csv";
        Path path = new Path(filePath);
        
        try (FSDataOutputStream out = fs.create(path)) {
            if (!group.isEmpty()) {
                // 写入CSV表头
                JsonNode firstNode = objectMapper.readTree(group.get(0));
                String header = String.join(",", firstNode.fieldNames());
                out.write((header + "\n").getBytes());
                
                // 写入每行数据
                for (String jsonRow : group) {
                    JsonNode node = objectMapper.readTree(jsonRow);
                    List<String> values = new ArrayList<>();
                    for (String field : firstNode.fieldNames()) {
                        values.add(node.get(field).asText());
                    }
                    String csvRow = String.join(",", values);
                    out.write((csvRow + "\n").getBytes());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        groupIdx++;
    }
}

方案2：给行打分组标识后用Dataset分区写入 #

如果要保留Spark写入API，可在分区内给每行打上(分区ID, 分组ID)标识，再转为Dataset按标识分区写入（无shuffle，分组逻辑在本地完成）：

import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.types.StructType;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import scala.Tuple2;

// 1. 给每行添加分区内分组标识
JavaPairRDD<String, Tuple2<Integer, Integer>> groupedRDD = stringRDD.mapPartitionsWithIndex((partitionId, rows) -> {
    long maxGroupSize = 1024 * 1024 * 10;
    long currentSize = 0;
    int currentGroupId = 0;
    List<Tuple2<String, Tuple2<Integer, Integer>>> result = new ArrayList<>();
    
    while (rows.hasNext()) {
        String row = rows.next();
        long rowSize = row.getBytes().length;
        if (currentSize + rowSize > maxGroupSize && currentSize > 0) {
            currentGroupId++;
            currentSize = 0;
        }
        result.add(new Tuple2<>(row, new Tuple2<>(partitionId, currentGroupId)));
        currentSize += rowSize;
    }
    return result.iterator();
}, false);

// 2. 提前定义Schema（需和JSON字段匹配）
StructType schema = new StructType()
    .add("col1", org.apache.spark.sql.types.DataTypes.StringType)
    .add("col2", org.apache.spark.sql.types.DataTypes.StringType)
    .add("partitionId", org.apache.spark.sql.types.DataTypes.IntegerType)
    .add("groupId", org.apache.spark.sql.types.DataTypes.IntegerType);

// 3. 转为Dataset并按标识分区写入
Dataset<Row> df = sparkSession.createDataset(groupedRDD.map(t -> {
    ObjectMapper mapper = new ObjectMapper();
    JsonNode node = mapper.readTree(t._1());
    return RowFactory.create(
        node.get("col1").asText(),
        node.get("col2").asText(),
        t._2()._1(),
        t._2()._2()
    );
}), org.apache.spark.sql.catalyst.encoders.RowEncoder.apply(schema));

df.write()
  .partitionBy("partitionId", "groupId")
  .format("csv")
  .option("header", "true")
  .option("maxRecordsPerFile", Integer.MAX_VALUE) // 确保每组一个文件
  .save("/home/myfilepath/");

关键注意点 #

• 禁止在Executor中使用SparkSession创建Dataset或执行SQL操作，此类操作仅能在Driver端完成
• 按大小分组时，需根据存储需求合理设置每组最大阈值
• 手动写文件时，必须通过分区ID+分组ID避免文件名冲突
• 使用Dataset写入时，必须提前定义Schema，不能依赖Spark自动推断（自动推断会触发Driver端处理，可能引发shuffle）

内容的提问来源于stack exchange，提问作者Nila