如何配置Spark Cassandra Connector连接多Cassandra集群？

你遇到的问题是因为Spark Thrift Server使用Cassandra Catalog时，若未明确指定默认Catalog，会尝试加载全局Cassandra配置（未设置时默认指向localhost）。以下是可行的解决办法：

一、修正Catalog配置并正确使用 #

你的第二种配置思路是对的，但需要注意两个关键点：

• 不要设置全局的spark.cassandra.*配置，避免默认连接干扰
• 查询时必须明确指定Catalog名称访问对应集群

修正后的启动命令：

./start-thriftserver.sh \
--master spark://master:7077 \
--hiveconf hive.server2.thrift.port=10000 \
--hiveconf hive.server2.thrift.bind.host=here \
--packages com.datastax.spark:spark-cassandra-connector_2.12:3.3.0 \
--conf spark.sql.extensions=com.datastax.spark.connector.CassandraSparkExtensions \
--conf spark.sql.catalog.db1=com.datastax.spark.connector.datasource.CassandraCatalog \
--conf spark.sql.catalog.db1.spark.cassandra.connection.host=db1 \
--conf spark.sql.catalog.db1.spark.cassandra.auth.username=user1 \
--conf spark.sql.catalog.db1.spark.cassandra.auth.password=pass1 \
--conf spark.sql.catalog.db2=com.datastax.spark.connector.datasource.CassandraCatalog \
--conf spark.sql.catalog.db2.spark.cassandra.connection.host=db2 \
--conf spark.sql.catalog.db2.spark.cassandra.auth.username=user2 \
--conf spark.sql.catalog.db2.spark.cassandra.auth.password=pass2 \
--conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hive.HiveCatalog

这里新增spark.sql.catalog.spark_catalog配置，将默认Catalog指定为Hive的，避免Spark自动尝试默认Cassandra连接。

之后在Thrift客户端执行查询时，必须指定Catalog、keyspace和表的完整路径：

-- 从db1集群读取数据
SELECT * FROM db1.your_keyspace.source_table;

-- 写入到db2集群
INSERT INTO db2.your_keyspace.target_table SELECT * FROM db1.your_keyspace.source_table;

二、代码方式手动配置（备选方案） #

如果Thrift Server的Catalog配置始终有问题，你可以通过编写Spark代码，分别为两个集群创建DataFrame直接完成迁移。

Scala示例： #

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .master("spark://master:7077")
  .appName("CassandraMigration")
  .config("spark.sql.extensions", "com.datastax.spark.connector.CassandraSparkExtensions")
  .getOrCreate()

// 读取db1集群的数据
val sourceDF = spark.read
  .format("org.apache.spark.sql.cassandra")
  .options(Map(
    "keyspace" -> "your_keyspace",
    "table" -> "source_table",
    "spark.cassandra.connection.host" -> "db1",
    "spark.cassandra.auth.username" -> "user1",
    "spark.cassandra.auth.password" -> "pass1"
  ))
  .load()

// 写入db2集群
sourceDF.write
  .format("org.apache.spark.sql.cassandra")
  .options(Map(
    "keyspace" -> "your_keyspace",
    "table" -> "target_table",
    "spark.cassandra.connection.host" -> "db2",
    "spark.cassandra.auth.username" -> "user2",
    "spark.cassandra.auth.password" -> "pass2"
  ))
  .mode("append")
  .save()

Python示例： #

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .master("spark://master:7077") \
    .appName("CassandraMigration") \
    .config("spark.sql.extensions", "com.datastax.spark.connector.CassandraSparkExtensions") \
    .getOrCreate()

# 读取db1集群数据
source_df = spark.read \
    .format("org.apache.spark.sql.cassandra") \
    .options(
        keyspace="your_keyspace",
        table="source_table",
        spark_cassandra_connection_host="db1",
        spark_cassandra_auth_username="user1",
        spark_cassandra_auth_password="pass1"
    ) \
    .load()

# 写入db2集群
source_df.write \
    .format("org.apache.spark.sql.cassandra") \
    .options(
        keyspace="your_keyspace",
        table="target_table",
        spark_cassandra_connection_host="db2",
        spark_cassandra_auth_username="user2",
        spark_cassandra_auth_password="pass2"
    ) \
    .mode("append") \
    .save()

关键注意点 #

• 确保spark-cassandra-connector版本与Spark、Cassandra版本兼容（你使用的3.3.0对应Spark 3.3.x）
• 若集群开启SSL，需添加对应SSL配置项（如spark.cassandra.connection.ssl.enabled等）

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