Spark Java Streaming创建Dataset时触发NullPointerException求助

这个空指针问题我之前也碰到过，核心原因是SparkSession是Driver端的对象，无法序列化传递到Executor节点执行，当你在RDD的foreach操作里直接调用Driver创建的spark.createDataset()时，Executor端的spark实例已经失效，导致sessionState为空，抛出NPE。

问题代码分析 #

你在directKafkaStream.foreachRDD()里的rdd.foreach(record -> {})是在Executor节点上执行的逻辑，但你直接引用了Driver初始化的spark对象——SparkSession并不实现Serializable接口，无法被序列化发送到Executor，所以在Executor端调用时会出现空指针。

另外，每条Kafka消息都创建一次Dataset的做法也非常低效，会带来大量的资源开销。

解决方案 #

下面给你两种可行的修改方案，根据你的数据量和场景选择：

方案1：Driver端批量处理（适合小数据量场景）

如果Kafka消息的吞吐量不大，可以先把RDD中的查询语句收集到Driver端，统一执行JDBC查询并创建Dataset：

directKafkaStream.foreachRDD(rdd -> {
    // 将RDD中的查询JSON收集到Driver端
    List<String> queryList = rdd.values().collect();
    SparkkafkaJson sk = new SparkkafkaJson();
    
    for (String queryJson : queryList) {
        List<String> jsonResult = sk.process_query(queryJson);
        // 在Driver端创建Dataset，这里的spark实例是有效的
        Dataset<String> resultDs = spark.createDataset(jsonResult, Encoders.STRING());
        System.out.println(resultDs.showString(10, 20));
        
        // 注册为临时表供后续使用
        resultDs.createOrReplaceTempView("presto_result");
        // 这里可以添加你的临时表操作逻辑
    }
});

方案2：Executor端分区处理（适合大数据量场景）

如果必须在Executor端处理，不要直接引用Driver的SparkSession，而是在每个分区内创建本地的SparkSession实例，并且尽量按分区批量处理，减少资源开销：

directKafkaStream.foreachRDD(rdd -> {
    rdd.foreachPartition(partition -> {
        // 每个分区创建一次SparkSession，复用资源
        SparkSession localSpark = SparkSession.builder()
                .config(SparkContext.getOrCreate().getConf())
                .getOrCreate();
        SparkkafkaJson sk = new SparkkafkaJson();
        
        while (partition.hasNext()) {
            String queryJson = partition.next()._2;
            List<String> jsonResult = sk.process_query(queryJson);
            Dataset<String> resultDs = localSpark.createDataset(jsonResult, Encoders.STRING());
            System.out.println(resultDs.showString(10, 20));
            
            // 注册临时表或执行其他操作
            resultDs.createOrReplaceTempView("presto_result_" + Thread.currentThread().getId());
        }
    });
});

额外优化建议 #

• JDBC连接池：你的process_query方法每次查询都创建新的JDBC连接，这会带来很大的性能损耗，建议使用连接池（比如HikariCP）来复用连接，提升查询效率。
• 资源关闭优化：在process_query的finally块中，确保ResultSet、Statement、Connection都被正确关闭，避免资源泄漏。
• Spark API规范：Spark 2.x之后推荐使用SparkSession，SQLContext可以通过spark.sqlContext()获取，不需要手动创建。

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