无法从 Spark 连接到 Cassandra（接触点包含多个数据中心）

Question

我正在尝试运行我的第一个 spark 作业（访问 Cassandra 的 Scala 作业）失败并显示以下错误：

java.io.IOException: Failed to open native connection to Cassandra at {<ip>}:9042
at com.datastax.spark.connector.cql.CassandraConnector$.com$datastax$spark$connector$cql$CassandraConnector$$createSession(CassandraConnector.scala:164)
at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$2.apply(CassandraConnector.scala:150)
at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$2.apply(CassandraConnector.scala:150)
at com.datastax.spark.connector.cql.RefCountedCache.createNewValueAndKeys(RefCountedCache.scala:31)
...........
............
Caused by: java.lang.IllegalArgumentException: Contact points contain multiple data centers: 
at com.datastax.spark.connector.cql.LocalNodeFirstLoadBalancingPolicy.init(LocalNodeFirstLoadBalancingPolicy.scala:47)
at com.datastax.driver.core.Cluster$Manager.init(Cluster.java:1099)
at com.datastax.driver.core.Cluster.getMetadata(Cluster.java:271)
at com.datastax.spark.connector.cql.CassandraConnector$.com$datastax$spark$connector$cql$CassandraConnector$$createSession(CassandraConnector.scala:157)

我们在这里做错了什么？

我正在使用：

• Spark 1.5.2
• Apache Cassandra 2.1.10
• spark-cassandra connector 1.3.1 /1.5.0-M2（两个连接器都试过了）
• Scala 版本 2.10.4

Answer 1

--> 根据作者的说法，正在进行一项解决此问题的工作。请参阅此答案下方的 cmets。

我在文档中找到了这个，希望对你有帮助：

override def init(cluster: Cluster, hosts: JCollection[Host]) {
    nodes = hosts.toSet
    // use explicitly set DC if available, otherwise see if all contact points have same DC
    // if so, use that DC; if not, throw an error
    dcToUse = localDC match { 
      case Some(local) => local
      case None => 
        val dcList = dcs(nodesInTheSameDC(contactPoints, hosts.toSet))
        if (dcList.size == 1) 
            dcList.head
        else 
            throw new IllegalArgumentException(s"Contact points contain multiple data centers: ${dcList.mkString(", ")}")
    }
    clusterMetadata = cluster.getMetadata
}

Answer 2

我在尝试使用 Apache Spark 2.x.x 连接两个 Cassandra 数据中心时遇到了同样的问题。

public class  SparkCassandraTest {  
  private static final String CASSANDRA_ENDPOINTS = "DC1_node1,DC1_node2,DC1_node3,DC2_node1,DC2_node2,DC2_node3";

  public static void main(String[] args) {
        sparkConf = new SparkConf().setAppName(APP_NAME);
        sparkConf.set("spark.cassandra.connection.host", CASSANDRA_ENDPOINTS);
        sparkConf.set("spark.cassandra.auth.username", CASSANDRA_USERNAME);
        sparkConf.set("spark.cassandra.auth.password", CASSANDRA_PASSWORD);
        sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //.....................
        //.....................
        //.....................
       }
}

原因：java.lang.IllegalArgumentException：要求失败：接触点包含多个数据中心：DC2-XXXXX2、DC1-XXXXX1

我通过连接任何一个 Cassandra 数据中心（DC1_node1、DC1_node2、DC1_node3）或（DC2_node1）来解决此问题强>、DC2_node2、DC2_node3）。