我必须为每个客户端每秒存储大约250个数值，这大约是每小时90万个数字。它可能不会是一整天的记录(可能一天5-10小时之间)，但是我将根据客户端ID和读取日期对数据进行分区。最大行长度约为22-23M，这仍然可以控制。永不过时，我的方案如下所示:
CREATE TABLE measurement (
clientid text,
date text,
event_time timestamp,
value int,
PRIMARY KEY ((clientid,date), event_time)
);
key 空间的复制因子为2，仅用于测试，特征码是
GossipingPropertyFileSnitch和NetworkTopologyStrategy。我知道复制因子3是更高的生产标准。接下来，我在公司的服务器，三台具有2个CPU x 2核心，16GB RAM和大量空间的裸机虚拟机上创建了一个小型集群。我和他们一起在千兆位局域网中。群集可运行，基于nodetool。
这是我用来测试设置的代码:
Cluster cluster = Cluster.builder()
.addContactPoint("192.168.1.100")
.addContactPoint("192.168.1.102")
.build();
Session session = cluster.connect();
DateTime time = DateTime.now();
BlockingQueue<BatchStatement> queryQueue = new ArrayBlockingQueue(50, true);
try {
ExecutorService pool = Executors.newFixedThreadPool(15); //changed the pool size also to throttle inserts
String insertQuery = "insert into keyspace.measurement (clientid,date,event_time,value) values (?, ?, ?, ?)";
PreparedStatement preparedStatement = session.prepare(insertQuery);
BatchStatement batch = new BatchStatement(BatchStatement.Type.LOGGED); //tried with unlogged also
//generating the entries
for (int i = 0; i < 900000; i++) { //900000 entries is an hour worth of measurements
time = time.plus(4); //4ms between each entry
BoundStatement bound = preparedStatement.bind("1", "2014-01-01", time.toDate(), 1); //value not important
batch.add(bound);
//The batch statement must have 65535 statements at most
if (batch.size() >= 65534) {
queryQueue.put(batch);
batch = new BatchStatement();
}
}
queryQueue.put(batch); //the last batch, perhaps shorter than 65535
//storing the data
System.out.println("Starting storing");
while (!queryQueue.isEmpty()) {
pool.execute(() -> {
try {
long threadId = Thread.currentThread().getId();
System.out.println("Started: " + threadId);
BatchStatement statement = queryQueue.take();
long start2 = System.currentTimeMillis();
session.execute(statement);
System.out.println("Finished " + threadId + ": " + (System.currentTimeMillis() - start2));
} catch (Exception ex) {
System.out.println(ex.toString());
}
});
}
pool.shutdown();
pool.awaitTermination(120,TimeUnit.SECONDS);
} catch (Exception ex) {
System.out.println(ex.toString());
} finally {
session.close();
cluster.close();
}
我通过阅读此处以及其他博客和网站上的帖子提出了代码。据我了解，客户端使用多个线程很重要，这就是为什么我要这样做。我也尝试使用异步操作。
最重要的结果是，无论我使用哪种方法，一批都会在5到6秒内执行，尽管可能最多需要10秒钟。如果我只输入一批(因此，仅〜65k列)，则结果相同。如果我使用的是愚蠢的单线程应用程序。老实说，我期望更多。尤其是因为我在具有本地实例的笔记本电脑上获得或多或少的相似性能。
第二个也许更重要的问题是我以不可预测的方式面临的异常。这两个:
和
最重要的是，我做错了什么吗？我应该重新组织加载数据的方式还是更改方案。我尝试减少行长(因此我有12小时的行)，但这并没有太大的区别。
=============================
更新:
我很粗鲁，忘了在回答问题后粘贴示例代码。它运行得相当不错，但是我将继续使用KairosDB进行研究，并使用Astyanax进行二进制传输。看起来我可以通过CQL获得更好的性能，尽管KairosDB在过载时可能会遇到一些问题(但我正在研究它)，而且Astyanax有点冗长，无法满足我的口味。不过，这是代码，我可能在某个地方弄错了。
超过5000时，信号灯插槽号对性能没有影响，它几乎是恒定的。
String insertQuery = "insert into keyspace.measurement (userid,time_by_hour,time,value) values (?, ?, ?, ?)";
PreparedStatement preparedStatement = session.prepare(insertQuery);
Semaphore semaphore = new Semaphore(15000);
System.out.println("Starting " + Thread.currentThread().getId());
DateTime time = DateTime.parse("2015-01-05T12:00:00");
//generating the entries
long start = System.currentTimeMillis();
for (int i = 0; i < 900000; i++) {
BoundStatement statement = preparedStatement.bind("User1", "2015-01-05:" + time.hourOfDay().get(), time.toDate(), 500); //value not important
semaphore.acquire();
ResultSetFuture resultSetFuture = session.executeAsync(statement);
Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
@Override
public void onSuccess(@Nullable com.datastax.driver.core.ResultSet resultSet) {
semaphore.release();
}
@Override
public void onFailure(Throwable throwable) {
System.out.println("Error: " + throwable.toString());
semaphore.release();
}
});
time = time.plus(4); //4ms between each entry
}
最佳答案
使用未记录的批处理有什么结果？您确定要使用批处理语句吗？
https://medium.com/@foundev/cassandra-batch-loading-without-the-batch-keyword-40f00e35e23e
关于java - Cassandra集群的插入性能和插入稳定性很差，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/27902232/