标签: datastax-java-driver

我从cassandra 2.0中提取了大量数据,但遗憾的是获得了超时异常.我的桌子:

CREATE KEYSPACE StatisticsKeyspace
  WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 3 };


CREATE TABLE StatisticsKeyspace.HourlyStatistics(
KeywordId text,
Date timestamp,
HourOfDay int,
Impressions int,
Clicks int,
AveragePosition double,
ConversionRate double,
AOV double,
AverageCPC double,
Cost double,
Bid double,
PRIMARY KEY(KeywordId, Date, HourOfDay)
);
CREATE INDEX ON StatisticsKeyspace.HourlyStatistics(Date);

我的查询:

SELECT KeywordId, Date, HourOfDay, Impressions, Clicks,AveragePosition,ConversionRate,AOV,AverageCPC,Bid 
FROM StatisticsKeyspace.hourlystatistics 
WHERE Date >= '2014-03-22' AND Date <= '2014-03-24'

我在我的cassandra.yaml文件中更改了配置.

read_request_timeout_in_ms: 60000
range_request_timeout_in_ms: 60000
write_request_timeout_in_ms: 40000
cas_contention_timeout_in_ms: 3000
truncate_request_timeout_in_ms: 60000 …

我有一个由 3 个 Cassandra 2.0 节点组成的集群。我的应用程序编写了一个测试，尝试将一些数据写入 Cassandra 或从 Cassandra 读取一些数据。一般来说，这工作得很好。

好奇的是，在我重新启动计算机后，此测试将失败，因为写入后我读取了之前写入的相同值，并且我得到的是 null 而不是该值，但写入时也不例外。\n如果我手动截断使用的列族，测试将通过。之后我可以按照我想要的频率执行这个测试，它一次又一次地通过。此外，Cassandra 中是否有值并不重要。结果总是一样的。

如果我查看 CLI 和 CQL-shell，会发现两种不同的视图：

有谁知道出了什么问题吗？CLI中的时间戳在重新执行后更新，所以这似乎是一个读取问题？

我的代码的一部分：\n对于我尝试过的插入

Insert.Options insert =   QueryBuilder.insertInto(KEYSPACE_NAME,TABLENAME)\n                .value(ID, id)\n                .value(JAHR, zonedDateTime.getYear())\n                .value(MONAT, zonedDateTime.getMonthValue())\n                .value(ZEITPUNKT, date)\n                .value(WERT, entry.getValue())\n                .using(timestamp(System.nanoTime() / 1000));\n

和

Insert insert = QueryBuilder.insertInto(KEYSPACE_NAME,TABLENAME)\n                .value(ID, id)\n                .value(JAHR, zonedDateTime.getYear())\n                .value(MONAT, zonedDateTime.getMonthValue())\n                .value(ZEITPUNKT, date)\n                .value(WERT, entry.getValue());\n

我的选择看起来像

Select.Where select = QueryBuilder.select(WERT)\n            .from(KEYSPACE_NAME,TABLENAME)\n            .where(eq(ID, id))\n            .and(eq(JAHR, zonedDateTime.getYear()))\n            .and(eq(MONAT, zonedDateTime.getMonthValue()))\n            .and(eq(ZEITPUNKT, Date.from(instant)));\n

一致性级别为 QUORUM（两者均适用）和复制因子 …

我无法对 Cassandra 节点进行查询。我能够建立与集群的连接并进行连接。但是，在执行查询时，它失败了

Caused by: com.datastax.driver.core.exceptions.NoHostAvailableException: All host(s) tried for query failed (no host was tried)
at com.datastax.driver.core.RequestHandler.reportNoMoreHosts(RequestHandler.java:217)
at com.datastax.driver.core.RequestHandler.access$1000(RequestHandler.java:44)
at com.datastax.driver.core.RequestHandler$SpeculativeExecution.sendRequest(RequestHandler.java:276)
at com.datastax.driver.core.RequestHandler.startNewExecution(RequestHandler.java:117)
at com.datastax.driver.core.RequestHandler.sendRequest(RequestHandler.java:93)
at com.datastax.driver.core.SessionManager.executeAsync(SessionManager.java:127)
... 3 more

这是我连接到集群的方式：

    List<String> servers = config.getCassandraServers();
    Builder builder = Cluster.builder();
    for (String server : servers) {
        builder.addContactPoints(server);
    }

    PoolingOptions opts = new PoolingOptions();
    opts.setCoreConnectionsPerHost(HostDistance.LOCAL, opts.getCoreConnectionsPerHost(HostDistance.LOCAL));

    // setup socket exceptions
    SocketOptions socketOpts = new SocketOptions();
    socketOpts.setReceiveBufferSize(1048576);
    socketOpts.setSendBufferSize(1048576);
    socketOpts.setTcpNoDelay(false);

    cluster = builder.withSocketOptions(socketOpts)
            .withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE)
            .withPoolingOptions(opts)
            .withReconnectionPolicy(new ConstantReconnectionPolicy(100L))
            .withLoadBalancingPolicy(new DCAwareRoundRobinPolicy(getColo(config)))
            .build();

    cluster.connect();

我正在使用带有 Datastax …

键空间和“人”表存在

你好，我是 Cassandra 的新手，所以我只是想从我的 Java 代码中执行选择查询，但我得到了一个异常，表不存在（但它存在。检查上面的链接）。请帮助我找出导致此问题的原因。先感谢您。

public class DS_Cassandra implements DS 
{
    Cluster cluster;
    Session session;
    private static final Logger log = Logger.getLogger( DS_Cassandra.class.getName() );

    public DS_Cassandra() 
    {
        cluster = Cluster.builder().addContactPoint("127.0.0.1").build();
        session = cluster.connect("testing");
    }

    @Override
    public ArrayList<Person> read() 
    {
        ArrayList<Person> list = new ArrayList<Person>();
        ResultSet results = session.execute("SELECT * FROM person"); //and here exception happened



2016-03-02 22:06:26 INFO  NettyUtil:83 - Did not find Netty's native epoll transport in the classpath, defaulting to NIO.
2016-03-02 22:06:26 INFO  DCAwareRoundRobinPolicy:95 - Using …

嗨，我刚刚在表 my_table 中添加了一个新列 Business_sys：

   ALTER TABLE my_table ALTER business_sys TYPE set<text>;

但是我再次删除了这个列名，因为我想更改列的类型：

  ALTER TABLE my_table DROP business_sys;

再次当我尝试添加不同类型的相同列名称时，我收到错误消息

"Cannnot add a collection with the name business_sys because the collection with the same name and different type has already been used in past"

我只是尝试执行此命令以添加不同类型的新列-

  ALTER TABLE my_table ADD business_sys list<text>;

我做错了什么？我对 Cassandra 很陌生。有什么建议？

所以在datastax doc 中，它指出 ConsistencyLevel 可以通过 QueryOptions 全局设置：

QueryOptions qo = new QueryOptions().setConsistencyLevel(ConsistencyLevel.ALL);

我注意到这不是一个静态方法并返回一个 QueryOptions 实例。这是否意味着简单地调用此方法不会全局设置默认一致性级别，并且在连接到 Cassandra 集群时我需要使用 QueryOptions？我的意思是，是否需要以下代码（QueryOption在构建Cluster对象时设置）

cluster = Cluster.builder().addContactPoint("192.168.0.30")
.withQueryOptions(new QueryOptions()
.setConsistencyLevel(ConsistencyLevel.ONE)
.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
.withLoadBalancingPolicy(new TokenAwarePolicy(new DCAwareRoundRobinPolicy()))
.build();
session = cluster.connect("demo");

我的问题是我无权访问构建 Cluster 实例的代码。这是否意味着我无法设置全局一致性级别而必须依赖于设置 per Statement？

我知道这nodetool describecluster可以为我提供有关 cassandra 中架构分歧的信息。但是，我想看看该信息是否可通过 JMX 获得。如果可用，有人可以向我指出该信息的完全限定包/指标名称吗？

我正在尝试使用 Prometheus 设置警报。

我正在使用 spark 2.4.1 版本和 java8 将数据复制到 cassandra-3.0。

我的火花作业脚本是

$SPARK_HOME/bin/spark-submit \
    --master yarn \
    --deploy-mode cluster \
    --name MyDriver  \
    --jars "/local/jars/*.jar" \
    --files hdfs://files/application-cloud-dev.properties,hdfs://files/column_family_condition.yml \
    --class com.sp.MyDriver \
    --executor-cores 3 \
    --executor-memory 9g \
    --num-executors 5 \
    --driver-cores 2 \
    --driver-memory 4g \
    --driver-java-options -Dconfig.file=./application-cloud-dev.properties \
    --conf spark.executor.extraJavaOptions=-Dconfig.file=./application-cloud-dev.properties \
    --conf spark.driver.extraClassPath=. \
    --driver-class-path . \
     ca-datamigration-0.0.1.jar application-cloud-dev.properties

以为工作成功了，我的日志文件中充满了以下 WARN。

WARN  org.apache.spark.storage.BlockManagerMasterEndpoint - No more replicas available for rdd_558_5026 !
2019-09-20 00:02:37,882 [dispatcher-event-loop-1] WARN    org.apache.spark.storage.BlockManagerMasterEndpoint - No more replicas available for …

在处理数据需要在 UUID 上排序的用例时，这些数据都是类型 1 或基于时间并使用 Datastax Cassandra Java 驱动程序库 (UUIDS.timebased()) 生成的，我发现 UUID.compareTo 没有对某些数据进行排序UUID 正确。compareTo 中的逻辑是

    /**
 * Compares this UUID with the specified UUID.
 *
 * <p> The first of two UUIDs is greater than the second if the most
 * significant field in which the UUIDs differ is greater for the first
 * UUID.
 *
 * @param  val
 *         {@code UUID} to which this {@code UUID} is to be compared
 *
 * @return  -1, 0 or 1 as this …

我正在尝试将 Kafka Streams 聚合结果插入到 Cassandra 中。但是我在 500 毫秒后连接期间出现超时错误。为什么会发生这种情况以及如何增加此超时或是否可以关闭此查询？

Caused by: com.datastax.oss.driver.api.core.DriverTimeoutException: query 'SELECT * FROM system.peers' timed out after PT0.5S
    at com.datastax.oss.driver.api.core.DriverTimeoutException.copy(DriverTimeoutException.java:34)
    at com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures.getUninterruptibly(CompletableFutures.java:149)
    at com.datastax.oss.driver.api.core.session.SessionBuilder.build(SessionBuilder.java:612)
    at aggregator.sink.CassandraSink.connect(CassandraSink.java:94)
    at aggregator.sink.CassandraSink.init(CassandraSink.java:62)
    at aggregator.sink.CassandraSink_ClientProxy.init(CassandraSink_ClientProxy.zig:253)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$init$0(ProcessorNode.java:97)
    at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.init(ProcessorNode.java:94)

datastax-java-driver 配置：

Caused by: com.datastax.oss.driver.api.core.DriverTimeoutException: query 'SELECT * FROM system.peers' timed out after PT0.5S
    at com.datastax.oss.driver.api.core.DriverTimeoutException.copy(DriverTimeoutException.java:34)
    at com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures.getUninterruptibly(CompletableFutures.java:149)
    at com.datastax.oss.driver.api.core.session.SessionBuilder.build(SessionBuilder.java:612)
    at aggregator.sink.CassandraSink.connect(CassandraSink.java:94)
    at aggregator.sink.CassandraSink.init(CassandraSink.java:62)
    at aggregator.sink.CassandraSink_ClientProxy.init(CassandraSink_ClientProxy.zig:253)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$init$0(ProcessorNode.java:97)
    at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.init(ProcessorNode.java:94)

datastax-java-driver.basic.contact-points.0=10.0.111.61:9042
datastax-java-driver.basic.load-balancing-policy.local-datacenter=datacenter1
datastax-java-driver.basic.session-keyspace=my_ks
datastax-java-driver.basic.request.timeout=20 seconds
datastax-java-driver.advanced.auth-provider.class=PlainTextAuthProvider
datastax-java-driver.advanced.auth-provider.username=${CASSANDRA_USR}
datastax-java-driver.advanced.auth-provider.password=${CASSANDRA_PWD}
datastax-java-driver.advanced.protocol.version=V4
datastax-java-driver.advanced.timestamp-generator.force-java-clock=true …