Spark流处理写入Cassandra报错及性能优化咨询

一、解决NoClassDefFoundError报错 #

这个错误的核心原因有两个，我帮你逐一排查解决：

1. Spark与Kafka连接器版本不兼容 #

org.apache.spark.sql.execution.streaming.Source$class找不到，本质是你的Spark版本和spark-sql-kafka-0-10连接器版本不匹配。Spark结构化流对Kafka连接器的版本要求非常严格，必须保证两者版本完全对齐。比如：

• 若使用Spark 2.4.x，Kafka连接器必须对应spark-sql-kafka-0-10_2.11:2.4.x（注意Scala版本也要匹配）
• 若使用Spark 3.x，连接器版本要同步升级到对应3.x系列

你需要检查项目依赖管理文件（Maven的pom.xml或SBT的build.sbt），确保Kafka连接器版本和Spark核心、Spark SQL的版本完全一致。

2. 写入流配置冲突 #

你的代码里同时设置了.foreach(writeToCassandra(cassandraConnector))和.format("console")，这是错误的——Spark Streaming的DataStreamWriter只能指定一种输出方式，不能同时输出到自定义ForeachWriter和控制台。你需要删除其中一个配置，比如测试阶段保留console，正式写入Cassandra时去掉console配置。

修改后的写入代码示例：

val query = dataStream
  .writeStream
  .outputMode(OutputMode.Append())
  .foreach(writeToCassandra(cassandraConnector))
  // 移除.format("console")这一行
  .start()

二、Cassandra写入性能优化（从6000条/秒提升） #

你当前用ForeachWriter逐行写入Cassandra是性能瓶颈的核心原因——Cassandra是分布式数据库，单条写入的网络和协调开销极大，必须用批量写入才能发挥它的性能优势。以下是具体优化方案：

1. 使用Spark Cassandra Connector原生批量写入 #

放弃自定义ForeachWriter逐行写入，直接用官方Cassandra连接器进行批量流写入，这是最高效的方式。示例代码如下：

import org.apache.spark.sql.cassandra._

// 先将Kafka的原始数据转换为Cassandra表对应的结构
// 假设Kafka消息是JSON格式，先解析成对应Schema
val yourSchema = // 定义与Cassandra表匹配的StructType
val cassandraReadyStream = dataStream
  .selectExpr("CAST(value AS STRING)")
  .from_json($"value", yourSchema)
  .select("col1", "col2", "col3") // 选择Cassandra表需要的字段

// 直接批量写入Cassandra
val query = cassandraReadyStream
  .writeStream
  .outputMode(OutputMode.Append())
  .format("org.apache.spark.sql.cassandra")
  .option("keyspace", "your_target_keyspace")
  .option("table", "your_target_table")
  .option("checkpointLocation", "/path/to/hdfs/checkpoint") // 必须设置checkpoint保证容错
  .start()

2. 调整关键配置参数 #

在你的SparkConf中添加以下Cassandra优化参数，大幅提升批量写入效率：

val conf = new SparkConf()
  // 你的原有配置...
  .set("spark.cassandra.output.batch.size.rows", "2000") // 每次批量写入行数，建议1000-5000
  .set("spark.cassandra.output.concurrent.writes", "12") // 每个executor的并发写入任务数
  .set("spark.cassandra.connection.connections_per_executor_max", "16") // 每个executor到Cassandra的最大连接数
  .set("spark.cassandra.output.batch.grouping.buffer.size", "10000") // 批量分组缓冲区大小
  .set("spark.cassandra.output.throughput_mb_per_sec", "100") // 单executor写入吞吐量限制，避免压垮Cassandra

3. 优化Spark并行度与Kafka消费配置 #

• Kafka消费端：设置maxOffsetsPerTrigger控制每次触发的消息拉取量，避免一次拉取过多数据导致Spark过载：

  val dataStream = spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "192.168.0.78:9092,192.168.0.78:9093,192.168.0.78:9094")
    .option("subscribe", "historyfleet")
    .option("maxOffsetsPerTrigger", "100000") // 每次触发拉取10万条，根据集群能力调整
    .load()
  

• Spark并行度：你当前配置9个executor（每个1核），总核数9。可以根据Cassandra集群的节点数调整，比如Cassandra是3节点且每个节点有4核，Spark总核数可设为12-16，保证和Cassandra的处理能力匹配。

4. 其他优化细节 #

• 确保Cassandra表的分区键设计合理，避免热点分区——如果所有写入数据集中在一个分区，Cassandra性能会急剧下降。
• 关闭Spark的speculation（将spark.speculation设为false），流处理中speculation会导致重复写入Cassandra，还会增加额外开销。
• 保证Kafka topic的分区数至少和Spark executor总核数一致，这样消费端才能并行拉取数据，避免消费瓶颈。

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