Spark Cassandra流写入未自动创建Keyspace问题求助

首先直接回应你的核心疑问：Spark Cassandra Connector（无论批处理还是流处理模式）默认不会自动创建Keyspace和表——这正是你在Flask查询时遇到cassandra.InvalidRequest: Error from server: code=2200 [Invalid query] message="Keyspace MyKeyspace does not exist"错误的根本原因。下面分步骤帮你排查和解决：

1. 开启自动创建Keyspace/表的配置 #

如果你希望Spark自动帮你创建Keyspace和表，需要显式添加两个关键配置：

• spark.cassandra.keyspace.create.enabled=true: 允许自动创建不存在的Keyspace
• spark.cassandra.createTable.enabled=true: 允许自动创建不存在的表

你可以选择全局配置（在SparkSession初始化时设置），或者在流写入代码中单独设置：

方式一：全局配置SparkSession #

val spark = SparkSession.builder()
  .appName("FlightRecommendationsStream")
  .config("spark.cassandra.connection.host", "cassandra")
  .config("spark.sql.extensions", "com.datastax.spark.connector.CassandraSparkExtensions")
  // 新增自动创建配置
  .config("spark.cassandra.keyspace.create.enabled", "true")
  .config("spark.cassandra.createTable.enabled", "true")
  .getOrCreate()

方式二：在流写入的foreachBatch中单独设置 #

val flightRecommendations = finalPredictions.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) => 
  batchDF 
    .write 
    .cassandraFormat("MytableName", "MyKeyspace") 
    .option("cluster", "cassandra_cluster")
    // 新增自动创建配置
    .option("spark.cassandra.keyspace.create.enabled", "true")
    .option("spark.cassandra.createTable.enabled", "true")
    .mode("append") 
    .save 
}.start()

2. 验证Spark与Cassandra的连接是否正常 #

你提到spark-submit无报错但Keyspace未创建，建议先验证连接有效性：

• 进入Spark容器，启动Spark Shell执行测试查询：

  spark.sql("SELECT keyspace_name FROM system_schema.keyspaces").show()
  

  如果能列出Cassandra默认的Keyspace（如system、system_schema），说明连接正常；如果报错，检查Docker网络配置：确保Spark和Cassandra容器在同一个Docker网络中，且Cassandra容器名称确实是cassandra（与你spark.cassandra.connection.host配置一致）。
• 查看Cassandra容器日志，确认是否收到Spark的连接请求：

  docker logs <cassandra-container-id>
  

3. 手动创建Keyspace/表（生产环境更推荐） #

自动创建虽然便捷，但可能因Schema映射问题导致表结构不符合预期（比如主键缺失、数据类型不匹配）。生产环境建议手动在Cassandra中创建：

-- 创建Keyspace（根据你的部署调整复制策略）
CREATE KEYSPACE IF NOT EXISTS MyKeyspace 
WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};

-- 创建表（根据你的finalPredictions DataFrame Schema定义字段和主键）
CREATE TABLE IF NOT EXISTS MyKeyspace.MytableName (
  flight_id INT,
  user_id INT,
  recommendation_score DOUBLE,
  created_at TIMESTAMP,
  PRIMARY KEY (flight_id, user_id) -- 示例复合主键，根据业务调整
);

手动创建后，你的流写入代码无需添加自动创建配置，直接写入即可。

额外注意点 #

• 确保你的finalPredictions DataFrame的字段名称、数据类型与Cassandra表的定义完全匹配，否则会写入失败。
• 流处理的foreachBatch模式下，每次批次都会执行写入逻辑，开启自动创建配置后Connector会自动处理重复创建的问题。

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