Spark将RDD保存至Cassandra时出现NoSuchElementException错误求助

先拆分你遇到的两个问题，逐个分析解决：

1. NoSuchElementException: key not found: date 错误 #

这个错误本质是Spark Cassandra Connector在尝试将RDD数据映射到Cassandra表时，找不到名为date的字段，常见原因和解决办法：

• 字段名不匹配或缺失：

  • 先检查你的Cassandra表结构，确认确实存在date列（注意Cassandra默认列名是小写的，如果你代码里用了Date或者DATE这种大写形式，会直接导致找不到）。
  • 再看从Kafka获取并处理后的RDD数据：你从Kafka拿到的lines是(String, String)类型的键值对吧？如果是的话，你需要把字符串解析成包含date字段的结构（比如Case Class或者Map），如果解析过程中漏掉了date，或者键名拼写错误，就会触发这个错误。
  • 举个正确的映射示例：假设你的Cassandra表是这样的：

    CREATE TABLE test.kafka_data (
        id UUID PRIMARY KEY,
        date timestamp,
        content text
    );
    

    那你的Spark代码里需要把Kafka数据转换成对应的Case Class：

    case class KafkaData(id: UUID, date: Timestamp, content: String)
    val processedData = lines.map { case (key, value) =>
        // 这里要正确解析value中的date字段，比如从JSON解析
        val json = JSON.parseFull(value).get.asInstanceOf[Map[String, Any]]
        KafkaData(UUID.randomUUID(), json("date").asInstanceOf[Timestamp], json("content").asInstanceOf[String])
    }
    processedData.saveToCassandra("test", "kafka_data")
    

    如果解析时json("date")不存在，或者Case Class字段名和Cassandra列名不匹配，就会报错。

• 映射关系错误：
  如果用Tuple来映射Cassandra表，要确保Tuple的元素顺序和Cassandra表的列顺序完全一致，并且包含date对应的位置。比如表列顺序是id, date, content，你的Tuple就得是(UUID, Timestamp, String)，不能缺项或者顺序错位。

2. Disconnected from Cassandra cluster: Test Cluster 警告 #

这个提示说明Spark和Cassandra的连接出现了中断，可能是连接配置或Cassandra服务本身的问题：

• 连接地址配置问题：
  你代码里设置的spark.cassandra.connection.host是127.0.1.1，先确认Cassandra是否监听这个地址。可以查看Cassandra配置文件cassandra.yaml里的listen_address和rpc_address，如果是127.0.0.1，那把配置改成127.0.0.1试试——有时候127.0.1.1是系统主机名的映射，Cassandra可能没绑定这个地址。

• Cassandra服务状态异常：
  先检查Cassandra是否正常运行，在终端执行：

  nodetool status
  

  如果提示nodetool: Failed to connect to '127.0.0.1:7199' - ConnectException: Connection refused，说明Cassandra没启动，先启动服务：

  sudo service cassandra start
  

• 配置未正确应用：
  注意SparkConf的set方法是返回新对象，如果你只是写了val cass=sparkConf.set("spark.cassandra.connection.host","127.0.1.1")，但创建StreamingContext时用的还是原来的sparkConf，那这个配置根本没生效。正确写法应该是：

  val sparkConf = new SparkConf()
    .setAppName("KafkaToCassandra")
    .setMaster("local[*]") // 本地测试用，生产环境去掉
    .set("spark.cassandra.connection.host", "127.0.0.1")
  val ssc = new StreamingContext(sparkConf, Seconds(5))
  

• 端口或认证问题：
  如果Cassandra修改了默认CQL端口（默认9042），需要添加配置：set("spark.cassandra.connection.port", "你的端口号")。如果Cassandra开启了用户名密码认证，还要加上：

  .set("spark.cassandra.auth.username", "your_username")
  .set("spark.cassandra.auth.password", "your_password")
  

额外建议 #

在保存数据到Cassandra之前，可以先打印RDD内容，确认数据结构是否正确，比如：

processedData.foreachRDD { rdd =>
    println("当前批次数据：")
    rdd.take(5).foreach(println)
    rdd.saveToCassandra("test", "kafka_data")
}

这样能快速排查数据是否包含date字段，以及结构是否和Cassandra表匹配。

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