cassandra java驱动程序连接抛出noHostAvailableException

Posted 2023-03-22

【中文标题】cassandra java驱动程序连接抛出noHostAvailableException

【英文标题】：cassandra java driver connection throwing noHostAvailableException

【发布时间】：2019-02-25 20:59:50

【问题描述】：

我有一个带有两个节点的 cassandra 集群.. 我已经设置了 spark 作业来从这个具有 3651568 个键的 cassandra 集群中进行查询。

import com.datastax.spark.connector.rdd.ReadConf
import org.apache.spark.sql.cassandra
import org.apache.spark.sql.SparkSession

val conf = new SparkConf(true).set("spark.cassandra.connection.host", "hostname)
val sc = new SparkContext(conf)

val spark = SparkSession.builder().master("local").appName("Spark_Cassandra").config("spark.cassandra.connection.host", "hostname").getOrCreate()
val studentsDF = spark.read.cassandraFormat("keyspacename", "tablename").options(ReadConf.SplitSizeInMBParam.option(32)).load()
studentsDF.show(1000)

我能够查询前 1000 行，但我无法找到从 1001th 行读取到第 2000 行的方法，以便我可以使用 spark 作业从 Cassandra 表中批量读取数据。

根据我开始使用 java 驱动程序的建议

这里是完整的解释

我必须使用 datastax java 驱动程序从 cassandra 数据库查询。我正在使用 datastax java 驱动程序版本cassandra-java-driver-3.5.1 和 apache-cassandra 版本apache-cassandra-3.0.9，我尝试通过安装 jar 来解决依赖关系我还检查了 yaml文件种子、listen_address、rpc_address 都指向我的主机并且 start_native_transport 设置为 true 这是我建立与 cassandra 数据库的连接的 java 代码 `

import java.net.InetAddress;
  import com.datastax.driver.core.Metadata;
  import java.net.UnknownHostException;
  import com.datastax.driver.core.Cluster;
  import com.datastax.driver.core.Cluster.Builder;
  import com.datastax.driver.core.Session;
  import com.datastax.driver.core.ResultSet;
  import com.datastax.driver.core.Row;
public class Started 
    public void connect()
    
     try
       
         Cluster cluster;
         Session session;
         cluster = Cluster.builder().addContactPoints("***.***.*.*").build();
       cluster.getConfiguration().getSocketOptions().setReadTimeoutMillis(2000);
         System.out.println("Connected to cluster:");
         session= cluster.connect("demo");
         Row row = session.execute("SELECT ename FROM demo.emp").one();
         System.out.println(row.getString("ename"));
         cluster.close();
        
          catch (Exception e) 
              e.printStackTrace();
              
           
    public static void main(String[] args)
     
       Started st = new Started();
       st.connect();
       
          

`

我在 cassandra 集群中只有一个节点，它已启动并正在运行。我也可以在 9042 端口上对它进行 cqlsh .. 到目前为止一切顺利，但是当我运行我的 java 程序时，我收到了这个错误或异常消息......

Connected to cluster:
`

com.datastax.driver.core.exceptions.NoHostAvailableException: All host(s) tried for query failed (tried: /***.***.*.*:9042 (com.datastax.driver.core.exceptions.TransportException: [/***.***.*.*:9042] Cannot connect))
            at com.datastax.driver.core.ControlConnection.reconnectInternal(ControlConnection.java:232)
            at com.datastax.driver.core.ControlConnection.connect(ControlConnection.java:79)
            at com.datastax.driver.core.Cluster$Manager.negotiateProtocolVersionAndConnect(Cluster.java:1631)
            at com.datastax.driver.core.Cluster$Manager.init(Cluster.java:1549)
            at com.datastax.driver.core.Cluster.init(Cluster.java:160)
            at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:342)
            at com.datastax.driver.core.Cluster.connect(Cluster.java:292)
            at Started.connect(Started.java:22)
            at Started.main(Started.java:34)

`

谁能帮忙！！

【参考方案1】：

这可能不适合 Spark。 Show 例如仅显示 1000 条记录，但不保证记录的顺序。多次调用可能会产生不同的结果。

如果您想在 Spark 中进行分页，您最好的选择可能是将结果作为本地迭代器获取，但同样这可能不是最好的处理方式。 Spark 是一个用于处理远程集群上的数据的系统。这意味着在数据框 api 中进行处理。

如果您真的只想慢慢翻​​阅记录，您可以使用toLocalIterator 将批次抓取回您的驱动程序机器（不推荐）。但是您可以通过使用 Java 驱动程序执行 Select (*) 来完成类似的操作。返回给您的结果集迭代器将在您浏览结果时自动对结果进行分页。

使用 Java 驱动程序分页的示例

https://docs.datastax.com/en/developer/java-driver/3.2/manual/paging/

ResultSet rs = session.execute("your query");
  for (Row row : rs) 
  // Process the row ...
  // By default this will only pull a new "page" of data from cassandra
  // when the previous page has been fully iterated through. See the
  // docs for more details    

使用 Spark 远程处理数据的示例

RDD Docs for Cassandra Dataframe Docs for Cassandra //RDD API sparkContext.cassandraTable("ks","tab").foreach(row => //processRow)

//Dataframe API - although similar foreach is available here as well
spark.read.format("org.apache.spark.sql.cassandra")
  .load()
  .select(//do some transforms)
  .write(//pickoutput of request)

使用到localIterator的例子，可能是最不相关的方法

Why you might want to do this with an example

// This reads all data in large blocks to executors, those blocks are then pulled one at a time back to the Spark Driver.
sparkContext.cassandraTable("ks","tab").toLocalIterator

【讨论】：

感谢您的友好回复。我还想知道使用 spark 作业从 cassandra 表中获取数十万条记录的最佳做法是什么，因为如果我将使用 select(*) 那么它会在使用 cqlsh 时抛出 java 堆错误或操作超时。 .

如果您使用驱动程序内部分页按上述方式分页记录，您应该不会遇到 timwouts 或堆问题。如果收集整个结果集而不是使用迭代器分页，您只会遇到问题

你能给我一个使用 toLocaliterator 执行读取操作的例子，这样我就可以翻阅记录。只有语法会做

我会将它们添加到问题中。

再次@surbhibohra，如果您认为 toLocalIterator 是必要的，我强烈建议您重新考虑您的应用程序。如果您使用 Spark，很可能有一种更有效的方式来完成您想做的事情。如果实际上不需要 Spark 来完成这项工作，那么正常的分页应该绰绰有余。

【参考方案2】：

这是驱动兼容性的问题。最初我使用的是 cassandra-java-driver-3.5.1 和 apache-cassandra-3.0.9 。 切换到 cassandra-java-driver-3.0.8 和 apache-cassandra-3.0.9 并安装一些 jar 文件：slf4j-log4j12-1.7.7.jar、log4j-1.2.17.jar、netty-all-4.0.39.Final.jar .. 对我来说很好:)