cassandra java驱动程序连接引发noHostAvailableException
我有一个带有两个节点的cassandra群集。。 我已经设置了spark作业,从这个cassandra集群中进行查询,该集群有3651568个密钥
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 job从Cassandra表中批量读取数据
根据推荐,我开始使用java驱动程序
以下是完整的解释
我必须使用datastax java驱动程序从cassandra数据库进行查询。。我正在使用datastax java驱动程序版本cassandra-java-driver-3.5.1
和apache cassandra版本apache-cassandra-3.0.9
,我已经尝试通过安装JAR来解决依赖关系。我还检查了yaml文件种子、侦听地址、rpc地址都指向我的主机,并且start_native_transport设置为true
下面是我的java代码,用于建立与cassandra数据库的连接
`
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是一个在远程集群上处理数据的系统。这意味着在dataframe api中进行处理
如果你真的只是想慢慢翻阅记录,你可以使用
toLocalIterator
将批抓取回你的驱动程序机器(不推荐)。但是,您可以通过使用Java驱动程序执行Select(*)来完成类似的任务。返回给您的结果集迭代器将在您浏览结果时自动翻页浏览结果使用Java驱动程序分页的示例
https://docs.datastax.com/en/developer/java-driver/3.2/manual/paging/
使用Spark远程处理数据的示例
RDD Docs for Cassandra Dataframe Docs for Cassandra //RDD API sparkContext。cassandraTable(“ks”、“tab”)。foreach(row=>;//processRow)
使用本地化器的示例,可能是最不相关的方法
Why you might want to do this with an example