Spark左外连接意外返回Optional.empty引发NPE问题求助

问题场景 #

在AWS Glue Spark 3.3.1环境中，尝试为节点对RDD的节点分配唯一ID：

1. 从节点对RDD中提取所有节点，去重后通过zipWithUniqueId生成ID映射RDD
2. 用左外连接将原节点对与ID映射合并，替换原节点为带ID的新节点
3. 运行时触发NPE，原因是row._2()._2().get()调用了空的Optional——逻辑上所有节点都应存在于ID映射RDD中，但左外连接未匹配到结果

核心代码如下：

JavaPairRDD<Node, Node> pairs = // ... 初始化逻辑
JavaPairRDD<Node, Long> index = pairs
        .flatMap(tuple -> Arrays.asList(tuple._1(), tuple._2()).iterator())
        .distinct()
        .zipWithUniqueId();

pairs.leftOuterJoin(index)
        .mapToPair(new MergeJoinResult())
        .mapToPair(Tuple2::swap)
        .leftOuterJoin(index)
        .mapToPair(new MergeJoinResult())
        .mapToPair(Tuple2::swap);

static class MergeJoinResult implements
        PairFunction<Tuple2<Node, Tuple2<Node, Optional<Long>>>, Node, Node>,
        Serializable {
    @Override
    public Tuple2<Node, Node> call(Tuple2<Node, Tuple2<Node, Optional<Long>>> row) throws Exception {
        return Tuple2.apply(new Node(row._1(), row._2()._2().get()), row._2()._1());
    }
}

已排查的信息 #

• 验证pairs和index RDD的数据均存在，无缺失
• Node类的equals/hashCode由Lombok生成，日志显示对象比较结果为true
• 发现index RDD按toString分组存在重复项，cogroup出现同键多条目，但这些对象的equals/hashCode结果一致
• 将Node序列化为JSON字符串作为连接键时，问题消失，指向序列化环节异常

根本原因分析 #

问题出在Kryo序列化导致Node对象在分布式传输后，equals/hashCode逻辑失效：
尽管Lombok生成的equals/hashCode逻辑本身正确，但Spark在分布式环境中，节点对象经过Kryo序列化反序列化后，内存中对象的hashCode可能与原始对象不一致，导致Shuffle阶段的分区、键匹配逻辑出错，原本应该匹配的键无法被正确关联，最终左外连接返回Optional.empty。

解决方案与排查建议 #

方案1：修复Kryo序列化配置 #

显式注册Node类到Kryo序列化器，避免默认序列化逻辑的不确定性：

1. 配置Spark使用Kryo并注册Node类

SparkConf conf = new SparkConf();
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
conf.registerKryoClasses(new Class[]{Node.class});
// AWS Glue环境下，传入配置初始化GlueContext
GlueContext glueContext = new GlueContext(new SparkContext(conf));

2. 若默认注册仍有问题，添加自定义Kryo序列化器

@Setter @Getter @EqualsAndHashCode
public class Node implements Serializable {
    private String nodeId; // 节点唯一标识字段

    // 自定义Kryo序列化器
    public static class NodeSerializer extends Serializer<Node> {
        @Override
        public void write(Kryo kryo, Output output, Node node) {
            output.writeString(node.getNodeId());
            // 写入其他字段
        }

        @Override
        public Node read(Kryo kryo, Input input, Class<Node> type) {
            Node node = new Node();
            node.setNodeId(input.readString());
            // 读取其他字段
            return node;
        }
    }
}

// 注册自定义序列化器
conf.getKryoRegistrator().register(Node.class, Node.NodeSerializer.class);

方案2：改用稳定的键类型 #

既然用JSON字符串作为键可以解决问题，可固化该方案，或直接使用Node的唯一标识字段（如nodeId）作为连接键，避免直接用自定义对象：

// 以节点唯一标识为键生成ID映射
JavaPairRDD<String, Long> index = pairs
        .flatMap(tuple -> Arrays.asList(tuple._1().getNodeId(), tuple._2().getNodeId()).iterator())
        .distinct()
        .zipWithUniqueId();

// 转换原节点对RDD为以唯一标识为键的结构，再进行连接
pairs.mapToPair(tuple -> Tuple2.apply(tuple._1().getNodeId(), tuple._2()))
        .leftOuterJoin(index)
        .mapToPair(row -> {
            String node1Id = row._1();
            Node node2 = row._2()._1();
            Long node1Index = row._2()._2().get();
            return Tuple2.apply(new Node(node1Id, node1Index), node2);
        })
        .mapToPair(Tuple2::swap)
        .mapToPair(tuple -> Tuple2.apply(tuple._1().getNodeId(), tuple._2()))
        .leftOuterJoin(index)
        .mapToPair(row -> {
            String node2Id = row._1();
            Node node1 = row._2()._1();
            Long node2Index = row._2()._2().get();
            return Tuple2.apply(node1, new Node(node2Id, node2Index));
        })
        .mapToPair(Tuple2::swap);

进一步排查建议 #

• 本地验证序列化前后的hashCode：将Node对象序列化后反序列化，比较前后的hashCode是否一致，确认Kryo的问题
• 统一分区器：显式指定HashPartitioner，确保连接的两个RDD分区逻辑一致

int numPartitions = pairs.getNumPartitions();
index = index.partitionBy(new HashPartitioner(numPartitions));
pairs = pairs.partitionBy(new HashPartitioner(numPartitions));

• 开启Shuffle调试日志：设置spark.logConf=true和spark.shuffle.spill.debug=true，查看Shuffle过程中键的分布，确认是否有同键被分到不同分区的情况

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