Spark中driver传值到worker出现null问题（含广播变量场景）

问题原因 #

这个问题是Java闭包序列化机制和Spark任务分发逻辑共同导致的：Spark分发任务时，会将算子内部的闭包逻辑+捕获的所有变量序列化后发送到worker节点。如果闭包捕获的是未符合要求的变量，序列化过程就会出现信息丢失，worker反序列化后拿到的值自然为null。

解决方案 #

1. 直接引用外部变量的场景修复

你需要确保传入闭包的变量是*显式final或实际不可修改（effectively final）*的局部变量，不要直接引用类的成员变量（会导致整个类实例被序列化，容易出现序列化失败），示例代码如下：

public static Dataset<SomeType> foo(Dataset<RSMRecommendation> dataset) {
    // 显式声明为final，确保可被闭包正确捕获序列化
    final String x = "test"; 
    Dataset<SomeType> ans =  dataset.map((MapFunction<SomeType, SomeType>) row -> {
        // 此处x可正常读取到值
    }, Encoders.bean(SomeType.class)).persist();
    return ans;
}

2. 广播变量的场景修复

你犯了广播变量使用的常见错误：不能在闭包内部直接调用Broadcast对象的value()方法。Broadcast本身是driver端的控制句柄，无法被正确序列化发送到worker，闭包捕获到的是序列化后的空壳对象，调用value()自然返回null。
正确的做法是在driver端（闭包外）提前取出广播值，赋值给局部final变量后再传入闭包：

public static Dataset<SomeType> foo(Dataset<RSMRecommendation> dataset, Broadcast<String> broadcastX) {
    // driver端提前取出广播值，赋值给显式final的局部变量
    final String x = broadcastX.value();
    Dataset<SomeType> ans =  dataset.map((MapFunction<SomeType, SomeType>) row -> {
        // 此处x可正常读取到值
    }, Encoders.bean(SomeType.class)).persist();
    return ans;
}

兜底兼容方案

如果你用的是Spark 2.x早期版本，存在lambda闭包捕获变量的已知bug，可以直接换成匿名内部类写法完全规避问题：

public static Dataset<SomeType> foo(Dataset<RSMRecommendation> dataset, Broadcast<String> broadcastX) {
    final String x = broadcastX.value();
    Dataset<SomeType> ans = dataset.map(new MapFunction<SomeType, SomeType>() {
        // 变量作为内部类的成员，确保序列化正常
        private final String innerX = x;
        @Override
        public SomeType call(SomeType row) throws Exception {
            // 直接使用innerX即可
        }
    }, Encoders.bean(SomeType.class)).persist();
    return ans;
}

额外排查点 #

如果上述方案仍无效，可以检查以下配置：

• 确认你没有在闭包里直接引用外部类的非静态成员变量，避免整个类实例序列化失败
• 如果你开启了Kryo序列化，确认已关闭spark.kryo.registrationRequired配置，或显式注册了用到的变量类型
• 确认自定义的实体类都实现了Serializable接口

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