将JSON格式Spark RDD写入Cassandra表失败求助

我来帮你搞定这个写入失败的问题！你的核心问题在于RDD里存的是JSON字符串，但Cassandra需要的是和表结构完全匹配的结构化数据，直接写肯定会报错。下面一步步给你拆解解决方案：

1. 先确保Spark-Cassandra连接器依赖正确 #

这是最容易踩的坑，连接器版本必须和你的Spark、Scala版本严格对应：

• 如果用spark-submit提交任务，要带上对应的包：

  spark-submit --packages com.datastax.spark:spark-cassandra-connector_2.12:3.4.1 你的应用jar包
  

  （比如Spark 3.4.x配3.4.1版本，Scala 2.12对应_2.12，根据你的实际版本调整）
• 如果是用SBT/Maven构建项目，直接在依赖里加入对应的坐标即可。

2. 把JSON字符串解析成结构化数据 #

你需要把RDD里的JSON字符串转换成Cassandra表对应的结构，推荐两种方式：

方式一：用Case Class + RDD解析 #

先定义和Cassandra表完全匹配的Case Class：

case class Station(id: String, date: String, temp: Int, press: Int)

然后用JSON解析库（比如json4s）把字符串转成Case Class实例：

import org.json4s._
import org.json4s.jackson.JsonMethods._

// 隐式解析规则
implicit val formats = DefaultFormats
// 把原始JSON RDD转成结构化的Station RDD
val parsedRDD = yourOriginalJsonRDD.map(jsonStr => parse(jsonStr).extract[Station])

方式二：用Spark SQL DataFrame（更推荐） #

直接把JSON RDD转成DataFrame，Spark会自动推断结构：

val df = spark.read.json(yourOriginalJsonRDD)
// 先打印结构确认和Cassandra表匹配
df.printSchema()

如果自动推断的类型有问题（比如temp被识别成String），可以手动指定Schema：

import org.apache.spark.sql.types._

val schema = StructType(Seq(
  StructField("id", StringType),
  StructField("date", StringType),
  StructField("temp", IntegerType),
  StructField("press", IntegerType)
))

val df = spark.read.schema(schema).json(yourOriginalJsonRDD)

3. 配置Cassandra连接参数 #

在代码里设置Cassandra的连接信息：

// 配置Cassandra节点地址
spark.conf.set("spark.cassandra.connection.host", "你的Cassandra节点IP")
// 默认端口是9042，如果改了就填实际端口
spark.conf.set("spark.cassandra.connection.port", "9042")
// 如果Cassandra开启了认证，还要加用户名密码
// spark.conf.set("spark.cassandra.auth.username", "your_username")
// spark.conf.set("spark.cassandra.auth.password", "your_password")

4. 写入Cassandra表 #

现在就可以把结构化数据写入Cassandra了，两种方式任选：

RDD方式 #

import com.datastax.spark.connector._

// 指定要写入的列（和表结构对应）
parsedRDD.saveToCassandra("mydata", "stations", SomeColumns("id", "date", "temp", "press"))

DataFrame方式（更推荐，容错性更好） #

df.write
  .format("org.apache.spark.sql.cassandra")
  .options(Map(
    "keyspace" -> "mydata",
    "table" -> "stations"
  ))
  .mode("append") // 按需选择：append追加，overwrite覆盖，ignore忽略
  .save()

最后检查几个常见坑 #

• 数据类型必须完全匹配：比如Cassandra里temp是int，解析后的数据不能是String，否则写入会失败
• 主键唯一性：你的表主键是id，如果重复写入相同id的数据，会覆盖旧数据，这是Cassandra的特性
• 网络连通性：确保Spark集群的所有节点都能访问到Cassandra的9042端口，防火墙不要拦截
• 连接器版本对应：一定要保证连接器版本和Spark、Scala版本兼容，不然会出现各种奇怪的报错

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