Spark Streaming从Twitter导数据入MongoDB遇clusterListener空指针异常

咱们先拆解你遇到的问题和代码里的几个关键问题，一步步来修复：

1. 核心错误根源 #

clusterListener can not be null这个异常主要来自两个问题：

• MongoDB连接配置缺失：你的Spark应用没有正确配置Mongo的连接信息，导致初始化连接时失败
• 违反Spark分布式运行规则：你在foreach（executor端执行的代码）里调用了sparkContext.parallelize——SparkContext只能在driver端使用，在executor端创建RDD会直接导致连接初始化异常

另外你代码里还有两个明显的低级失误：

• 所有Document的value都用了record(1)，这是复制粘贴时的错误，应该对应Array[String]里的不同索引
• 调用rdd.collect.foreach会把整个RDD的数据拉到driver端处理，完全浪费了Spark的分布式优势，数据量大时还会直接内存溢出

2. 分步修复方案 #

第一步：配置MongoDB连接信息 #

在Spark应用初始化阶段，必须指定带认证的Mongo连接URI（你的Mongo设置了用户名密码），示例代码如下：

val spark = SparkSession.builder()
  .appName("TwitterStreamToMongo")
  .master("local[*]") // 生产环境请移除这个配置
  .config("spark.mongodb.uri", "mongodb://admin:pwd@localhost:27017/twitter_db.tweets?authSource=admin")
  .getOrCreate()

这里twitter_db是你要使用的数据库名，tweets是集合名，可根据自己的需求修改。

第二步：重构写入Mongo的代码 #

彻底改掉你当前的foreachRDD逻辑，直接在RDD层面完成转换和写入，避免driver端的低效操作：

wordsArrays.foreachRDD { rdd =>
  // 先过滤掉长度不足的无效数据（你需要用到7个索引值）
  val validRecords = rdd.filter(_.length >= 7)
  // 将每条记录转换为Document类型的RDD
  val docRdd = validRecords.map { record =>
    new Document()
      .append("tweetId", record(0))
      .append("text", record(1))
      .append("favoriteCount", record(2))
      .append("retweetCount", record(3))
      .append("geoLocation", record(4))
      .append("language", record(5))
      .append("createdAt", record(6))
  }
  // 分布式写入MongoDB
  MongoSpark.save(docRdd)
}

这样做的优势：

• 所有操作都在分布式RDD层面执行，不会把数据拉到driver端
• 避开了在executor端使用SparkContext的错误
• 修复了索引引用错误的问题

第三步：清理依赖冲突 #

你的Spark版本是2.2.0，对应的mongo-spark-connector2.2.2是兼容的，但注意不要同时引入mongo-scala-driver——连接器已经包含了必要的驱动依赖，重复引入会导致版本冲突，建议从build.sbt里移除这个依赖：

// 移除这一行
"org.mongodb.scala" %% "mongo-scala-driver" % "2.2.0"

第四步：验证容器可达性 #

确保你的Spark应用能访问到本地的Mongo容器：

• 本地运行Spark的话，localhost:27017是可用的
• 如果Spark也运行在容器里，需要把连接URI里的localhost换成容器名mongo

3. 额外生产环境建议 #

• 不要在foreachRDD里执行任何driver端操作（比如创建RDD、初始化连接池），分布式逻辑都要放在map/filter等RDD算子中
• 若需要提升写入性能，可以用foreachPartition为每个分区创建一个Mongo连接，减少连接开销
• 生产环境务必移除master("local[*]")配置，提交到Spark集群运行

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