使用.NET Core 3.1在Spark中读取Kafka数据时遭遇NullPointerException问题求助

我之前也碰到过一模一样的空指针问题，结合你的异常栈、代码和环境信息来看，这个问题核心出在Spark处理Kafka认证配置的逻辑上，下面是具体的分析和可行的解决方案：

异常根源分析 #

从你给出的栈信息来看，空指针发生在KafkaConfigUpdater.scala:60，对应Spark 3.1.2的源码，这一行是在尝试读取SASL认证的jaas.config配置时，发现预期的配置项不存在或为null。哪怕你没有主动设置认证参数，如果Spark的全局配置里残留了认证相关项，或者你的Kafka Broker实际启用了认证但代码没配置，都会触发这个错误。

具体解决方案 #

1. 强制指定无认证连接（适用于无认证的Kafka环境） #

如果你的本地Kafka或者外部Broker没开认证，那可以在代码里显式添加security.protocol选项，强制Spark使用无认证模式，避免它自动尝试加载不存在的认证配置：

var stream = spark.ReadStream()
    .Format("kafka")
    .Option("kafka.bootstrap.servers", "127.0.0.1:9093")
    .Option("subscribe", "spark-input")
    .Option("startingOffsets", "earliest")
    .Option("failOnDataLoss", "false")
    .Option("kafka.security.protocol", "PLAINTEXT"); // 新增这一行

2. 补全SASL认证参数（适用于带认证的Kafka环境） #

如果你的外部Broker启用了SASL认证，必须完整配置所有必要的认证参数，不能遗漏。比如用PLAIN机制的配置示例：

var stream = spark.ReadStream()
    .Format("kafka")
    .Option("kafka.bootstrap.servers", "your-external-broker:9092")
    .Option("subscribe", "spark-input")
    .Option("startingOffsets", "earliest")
    .Option("failOnDataLoss", "false")
    .Option("kafka.security.protocol", "SASL_PLAINTEXT")
    .Option("kafka.sasl.mechanism", "PLAIN")
    .Option("kafka.sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"your-user\" password=\"your-pass\";");

注意：sasl.jaas.config的格式要严格符合Kafka要求，用户名和密码替换成你实际的认证信息。

3. 清除Spark全局配置的干扰 #

有时候spark-defaults.conf里可能配置了全局的Kafka认证参数，比如spark.sql.kafka.sasl.jaas.config，这会影响所有Kafka连接任务。可以在spark-submit启动时临时覆盖这些配置：

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.2 \
    --conf spark.sql.kafka.security.protocol=PLAINTEXT \
    --class org.apache.spark.deploy.dotnet.DotnetRunner \
    --master local \
    bin\Debug\netcoreapp3.1\microsoft-spark-3-1_2.12-2.0.0.jar \
    dotnet bin\Debug\netcoreapp3.1\SparkTest.dll

4. 修复代码中的流处理终止问题 #

你的代码里在WriteStream.Start()之后直接调用spark.Stop()，这会导致SparkSession立即停止，流处理根本没机会运行。应该改成等待流处理完成：

var query = dataFrame.WriteStream()
    .Format("console")
    .Start();
query.AwaitTermination(); // 等待流处理结束，程序会一直运行直到你手动终止
// spark.Stop(); // 不需要手动调用，AwaitTermination结束后会自动清理

5. 验证Kafka的可访问性 #

最后建议用Kafka自带的工具验证Broker和主题的可访问性，排除Kafka本身的问题：

# 列出Broker上的所有主题
kafka-topics.sh --list --bootstrap-server 127.0.0.1:9093

# 发送测试消息到目标主题
kafka-console-producer.sh --broker-list 127.0.0.1:9093 --topic spark-input

总结 #

优先尝试方案1（强制无认证）或者方案2（补全认证参数），这两个是解决这个空指针最直接的办法。同时记得修复代码中提前停止SparkSession的问题，否则流处理无法正常运行。

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