Scala从Spark Streaming的RDD中提取CassandraRow的技术问询

看来你已经搞定了Kafka数据流的接入和JSON解析到Case Class的环节，接下来我们可以通过两种方式完成到CassandraRow的转换（或者更高效的直接写入Cassandra）——推荐用Spark Cassandra Connector的DataFrame API，简洁又不容易出错；如果一定要手动构造CassandraRow，我也会给出对应的写法。

1. 先准备好依赖 #

首先确保你的项目引入了必要的依赖（以SBT为例）：

// Spark Cassandra Connector（版本要和你的Spark版本匹配，比如Spark 2.4.x用2.4.x系列）
libraryDependencies += "com.datastax.spark" %% "spark-cassandra-connector" % "2.5.2" % Provided

// JSON解析用的json4s（和你现有代码里的parse/extract对应）
libraryDependencies += "org.json4s" %% "json4s-native" % "3.6.12"

2. 定义Case Class与Cassandra表结构 #

先确保你的Case Class字段和Cassandra表的字段、类型完全对应。比如假设你的Cassandra表是这样的：

CREATE TABLE IF NOT EXISTS my_keyspace.history_events (
    gpsdt TIMESTAMP PRIMARY KEY,
    latitude DOUBLE,
    longitude DOUBLE,
    // 其他业务字段...
);

对应的Scala Case Class就要写成：

case class HistoryEvent(
    gpsdt: String, // JSON里是字符串格式的时间，后续转成Cassandra的TIMESTAMP
    latitude: Double,
    longitude: Double,
    // 补充其他需要的业务字段
)

3. 完整代码实现 #

这里修正了你原代码里的几个问题（比如避免用collect()拉数据到Driver端），同时提供两种实现方式：

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.json4s.DefaultFormats
import org.json4s.native.JsonMethods.parse
import org.apache.spark.sql.{SaveMode, SQLContext}
import org.apache.spark.sql.functions.{col, to_timestamp}
import com.datastax.spark.connector.CassandraRow
import java.sql.Timestamp

// 定义和Cassandra表匹配的Case Class
case class HistoryEvent(
    gpsdt: String,
    latitude: Double,
    longitude: Double
)

object KafkaToCassandraProcessor {
  def main(args: Array[String]): Unit = {
    // Spark配置，注意Cassandra连接信息
    val conf = new SparkConf()
      .setAppName("KafkaToCassandra")
      .setMaster("local[*]") // 生产环境请删除这行，用集群模式
      .set("spark.cassandra.connection.host", "你的Cassandra节点地址")

    val ssc = new StreamingContext(conf, Seconds(5))
    val sqlContext = new SQLContext(ssc.sparkContext)

    // Kafka连接参数
    val kafkaParams = Map(
      "metadata.broker.list" -> "你的Kafka Broker地址",
      "group.id" -> "kafka-cassandra-consumer-group",
      "auto.offset.reset" -> "latest"
    )
    val topics = Set("你的Kafka Topic名称")

    // 创建Kafka Direct Stream
    val kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
    // 提取Kafka消息体
    val jsonRecords = kafkaStream.map(_._2)

    jsonRecords.foreachRDD { rdd =>
      // 跳过空RDD，避免无效操作
      if (!rdd.isEmpty()) {
        implicit val formats = DefaultFormats
        import sqlContext.implicits._

        // 第一步：解析JSON到Case Class
        val eventRDD = rdd.map { jsonStr =>
          val jValue = parse(jsonStr)
          jValue.extract[HistoryEvent]
        }

        // --------------------------
        // 方式1：推荐用DataFrame写入Cassandra（最简洁高效）
        // --------------------------
        val eventDF = eventRDD.toDF()
        // 把字符串格式的时间转成Cassandra支持的Timestamp类型
        val formattedDF = eventDF.withColumn("gpsdt", to_timestamp(col("gpsdt"), "yyyy-MM-dd HH:mm:ss"))

        formattedDF.write
          .format("org.apache.spark.sql.cassandra")
          .options(Map(
            "keyspace" -> "my_keyspace",
            "table" -> "history_events"
          ))
          .mode(SaveMode.Append)
          .save()

        // --------------------------
        // 方式2：手动构造CassandraRow（如果需要直接操作Row对象）
        // --------------------------
        val cassandraRowRDD = eventRDD.map { event =>
          // 将字符串时间转成java.sql.Timestamp，匹配Cassandra的TIMESTAMP类型
          val timestamp = Timestamp.valueOf(event.gpsdt)
          // 按Cassandra表的字段顺序构造Row
          CassandraRow(
            timestamp,
            event.latitude,
            event.longitude
            // 其他字段按顺序添加
          )
        }

        // 如果需要用CassandraConnector批量写入，也可以这样（但DataFrame方式更推荐）
        // import com.datastax.spark.connector._
        // cassandraRowRDD.saveToCassandra("my_keyspace", "history_events")
      }
    }

    ssc.start()
    ssc.awaitTermination()
  }
}

4. 关键注意事项 #

• 绝对不要用collect()：你原代码里的rdd.collect().foreach会把整个RDD的数据拉到Driver节点，数据量大的时候直接OOM，一定要在Executor端用map或foreachPartition处理数据。
• 版本匹配：Spark Cassandra Connector的版本必须和你的Spark版本对应（比如Spark 2.4.x对应connector 2.4.x，Spark 3.x对应connector 3.x），否则会出现兼容性问题。
• 时间格式要一致：to_timestamp里的格式字符串要和JSON中的时间字符串完全匹配，比如JSON里是2017-04-12 00:25:10，就用yyyy-MM-dd HH:mm:ss，格式不匹配会导致时间解析失败。
• Cassandra连接配置：如果你的Cassandra集群有认证，还要在SparkConf里添加spark.cassandra.auth.username和spark.cassandra.auth.password参数。

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