我的版本：spark-2.1.1-bin-hadoop2.7，kafka_2.11-0.9.0.0，apache-cassandra-3.9。
val sparkConf = new SparkConf().setAppName("KafkaSparkStreaming").set("spark.cassandra.connection.host", "127.0.0.1")
val ssc = new StreamingContext(sparkConf, Seconds(5))
val topicpMap = "mytopic".split(",").map((_, 1.toInt)).toMap
val lines = KafkaUtils.createStream(ssc, "localhost:2181", "sparkgroup", topicpMap).map(_._2)
lines.map(line => { val arr = line.split(","); (arr(0),arr(1),arr(2),arr(3),arr(4)) }).saveToCassandra("sparkdata", "cust_data", SomeColumns("fname", "lname","url","product","cnt"))我试图运行下面的命令，但我得到了错误：
lines.map(line => {瓦尔arr = line.split（“，”）;（1），（2），（3），（4）}）.saveToCassandra（“sparkdata”，“cust_data”，“SomeColumns（“fname”，“lname”，“url”，“product”，“cnt”）
the error i am getting:
java.io.IOException: Failed to open native connection to Cassandra at {127.0.0.1}:9042
at com.datastax.spark.connector.cql.CassandraConnector$.com$datastax$spark$connector$cql$CassandraConnector$$createSession(CassandraConnector.scala:168)
at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$8.apply(CassandraConnector.scala:154)
at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$8.apply(CassandraConnector.scala:154)
at com.datastax.spark.connector.cql.RefCountedCache.createNewValueAndKeys(RefCountedCache.scala:32)
at com.datastax.spark.connector.cql.RefCountedCache.syncAcquire(RefCountedCache.scala:69)
at com.datastax.spark.connector.cql.RefCountedCache.acquire(RefCountedCache.scala:57)
at com.datastax.spark.connector.cql.CassandraConnector.openSession(CassandraConnector.scala:79)
at com.datastax.spark.connector.cql.CassandraConnector.withSessionDo(CassandraConnector.scala:111)
at com.datastax.spark.connector.cql.CassandraConnector.withClusterDo(CassandraConnector.scala:122)
at com.datastax.spark.connector.cql.Schema$.fromCassandra(Schema.scala:330)
at com.datastax.spark.connector.cql.Schema$.tableFromCassandra(Schema.scala:350)
at com.datastax.spark.connector.writer.TableWriter$.apply(TableWriter.scala:336)
at com.datastax.spark.connector.streaming.DStreamFunctions.saveToCassandra(DStreamFunctions.scala:53)
... 58 elided
Caused by: com.datastax.driver.core.exceptions.NoHostAvailableException: All host(s) tried for query failed (tried: /127.0.0.1:9042 (com.datastax.driver.core.exceptions.TransportException: [/127.0.0.1:9042] Cannot connect))
at com.datastax.driver.core.ControlConnection.reconnectInternal(ControlConnection.java:233)
at com.datastax.driver.core.ControlConnection.connect(ControlConnection.java:79)
at com.datastax.driver.core.Cluster$Manager.init(Cluster.java:1483)
at com.datastax.driver.core.Cluster.getMetadata(Cluster.java:399)
at com.datastax.spark.connector.cql.CassandraConnector$.com$datastax$spark$connector$cql$CassandraConnector$$createSession(CassandraConnector.scala:161)
... 70 more有人指出我的错误来解决吗？
1条答案
按热度按时间avwztpqn1#
NoHostAvailableException表示无法建立连接。这是一个较低的层，所以在尝试任何身份验证之前。因此，您的问题是连通性。以下是一些需要检查的事项：
1.您确定9042是正确的端口吗？是否更改了默认的本机端口？
1.有防火墙吗？这似乎不太可能，因为您正在尝试连接到127.0.0.1：9042。
1.您是否验证了Cassandra正在侦听端口9042（netstat）并且该端口已打开
无论哪种方式，这都是一个连接问题。