Spark DataFrame操作时触发NullPointerException问题排查

从你的描述和异常栈信息来看，这个问题的核心是Yarn集群模式下，Executor端执行任务时SparkSession相关实例出现了空引用——报错栈里的Caused by: java.lang.NullPointerException at org.apache.spark.sql.SparkSession$$anonfun$3.apply(SparkSession.scala:469)直接指向了这一点。而且本地模式正常、Yarn模式出问题，大概率是分布式环境下的序列化或者上下文传递问题。

可能的原因拆解 #

1. RDD元素携带了不可序列化的Driver端上下文引用 #

如果你的RDD里的元素，直接或间接引用了SparkSession、HiveContext这类Driver端的上下文对象，在Yarn模式下，这些对象需要序列化后发送到Executor，但SparkSession本身是不可序列化的，Executor端拿到的就是null引用，执行时自然触发NPE。

2. Schema定义存在隐性问题 #

虽然你能正常打印Schema，但如果Schema里包含了非序列化的自定义类型，或者Schema的元数据本身存在null字段，在Executor端解析数据和Schema匹配时就会抛出空指针。

3. Yarn环境的类路径/版本冲突 #

本地模式和Yarn集群的类路径、Spark版本如果不一致，或者存在Jar包冲突，会导致Executor端加载类时出现异常，间接引发NPE。比如你的应用用Spark 2.4.5打包，但集群是Spark 3.0，就可能出现这类兼容性问题。

具体排查与修复步骤 #

第一步：清理RDD中的上下文依赖 #

检查你生成RDD的逻辑，确保RDD的元素完全不依赖Driver端的SparkSession、HiveContext。比如如果你的RDD是从某个自定义类生成的，要确认这个类没有持有SparkSession的引用，并且实现了Serializable接口。

举个反例和修正示例：

// 错误示例：持有SparkSession引用，不可序列化
public class MyData implements Serializable {
    private SparkSession spark; // 这个会导致序列化问题
    private String value;
    // ...
}

// 正确示例：只包含业务数据
public class MyData implements Serializable {
    private String value;
    // ...
}

第二步：验证Schema的合法性 #

手动构建一个极简的Schema测试，比如只包含几个基本类型字段，替换原来的data.getSchema()，看是否还会触发异常：

import org.apache.spark.sql.types.*;

StructType testSchema = new StructType()
    .add(new StructField("id", IntegerType, false, Metadata.empty()))
    .add(new StructField("name", StringType, false, Metadata.empty()));

Dataset<Row> rows = applicationSession.getSparkSession()
    .createDataFrame(rdd, testSchema)
    .toDF("id", "name");

如果测试Schema能正常运行，说明原来的Schema存在问题，需要排查data.getSchema()的生成逻辑，确保所有字段类型都是Spark内置的可序列化类型。

第三步：优化序列化配置 #

切换到Kryo序列化器，它比默认的Java序列化兼容性更好，能处理更多复杂类型：

SparkSession spark = applicationSession.getSparkSession();
spark.conf().set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
// 如果有自定义类，注册到Kryo
spark.sparkContext().getConf().registerKryoClasses(new Class[]{MyData.class});

第四步：排查Yarn环境兼容性 #

• 确认你的应用打包用的Spark版本和Yarn集群的Spark版本完全一致，大版本小版本都要匹配。
• 用spark-submit --verbose命令提交应用，查看类路径加载情况，检查是否有重复的Jar包或者缺失的依赖。
• 如果有Hive相关操作，确认spark.sql.hive.metastore.version、spark.sql.hive.metastore.jars等配置和集群一致。

另外，你可以在RDD的map操作里加日志，打印每个元素的内容，确认是否有null元素或者异常数据导致问题——虽然你说RDD count是35，但不排除个别元素有问题。

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