【发布时间】：2014-03-13 18:33:50
【问题描述】：
我正在评估 cassandra。我正在使用 datastax 驱动程序和 CQL。
我想存储一些具有以下内部结构的数据，其中每次更新的名称都不同。
+-------+-------+-------+-------+-------+-------+
| | name1 | name2 | name3 | ... | nameN |
| time +-------+-------+-------+-------+-------+
| | val1 | val2 | val3 | ... | valN |
+-------+-------+-------+-------|-------+-------+
所以时间应该是列键，名称应该是行键。我用来创建这个表的 CQL 语句是：
CREATE TABLE IF NOT EXISTS test.wide (
time varchar,
name varchar,
value varchar,
PRIMARY KEY (time,name))
WITH COMPACT STORAGE
我希望架构是这种方式以便于查询。我还必须偶尔存储超过 65000 行的更新。所以使用 cassandra list/set/map 数据类型不是一个选项。
我必须能够每秒处理至少 1000 次宽行插入，并且名称/值对的数量虽然不同但很大 (~1000)。
问题如下：我编写了一个简单的基准测试，它执行 1000 个宽行插入，每个行插入 10000 个名称/值对。我使用 CQL 和 datastax 驱动程序的性能非常缓慢，而不使用 CQL（使用 astyanax）的版本在同一测试集群上具有良好的性能。
我已阅读此related question，并且在此问题的已接受答案中建议您应该能够通过使用批量准备语句，在 cassandra 2 中可用。
所以我尝试使用这些，但性能仍然很慢（对于在 localhost 上运行的小型三节点集群，每秒插入两次）。我是否遗漏了一些明显的东西，或者我必须使用较低级别的节俭 API？ 我在 astyanax 中使用 ColumnListMutation 实现了相同的插入，我每秒插入大约 30 次。
如果我必须使用较低级别的 thrift API：
它实际上已被弃用，还是因为它的级别较低而使用不方便？
我能否使用 CQL 查询使用 thrift api 创建的表？
以下是 scala 中的自包含代码示例。它只是创建一个批处理语句，用于插入具有 10000 列的宽行并重复计算插入性能。
我使用了 BatchStatement 选项和一致性级别，但没有什么能让我获得更好的性能。
我唯一的解释是，尽管批处理由准备好的语句组成，但条目被一个接一个地添加到行中。
package cassandra
import com.datastax.driver.core._
object CassandraTestMinimized extends App {
val keyspace = "test"
val table = "wide"
val tableName = s"$keyspace.$table"
def createKeyspace = s"""
CREATE KEYSPACE IF NOT EXISTS ${keyspace}
WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }
"""
def createWideTable = s"""
CREATE TABLE IF NOT EXISTS ${tableName} (
time varchar,
name varchar,
value varchar,
PRIMARY KEY (time,name))
WITH COMPACT STORAGE
"""
def writeTimeNameValue(time: String) = s"""
INSERT INTO ${tableName} (time, name, value)
VALUES ('$time', ?, ?)
"""
val cluster = Cluster.builder.addContactPoints("127.0.0.1").build
val session = cluster.connect()
session.execute(createKeyspace)
session.execute(createWideTable)
for(i<-0 until 1000) {
val entries =
for {
i <- 0 until 10000
name = i.toString
value = name
} yield name -> value
val batchPreparedStatement = writeMap(i, entries)
val t0 = System.nanoTime()
session.execute(batchPreparedStatement)
val dt = System.nanoTime() - t0
println(i + " " + (dt/1.0e9))
}
def writeMap(time: Long, update: Seq[(String, String)]) : BatchStatement = {
val template = session
.prepare(writeTimeNameValue(time.toString))
.setConsistencyLevel(ConsistencyLevel.ONE)
val batch = new BatchStatement(BatchStatement.Type.UNLOGGED)
for ((k, v) <- update)
batch.add(template.bind(k, v))
batch
}
}
这里是 astyanax 代码（从 astyanax example 修改），它在本质上将相同的事情快 15 倍。请注意，这也不使用异步调用，因此这是一个公平的比较。这要求列族已经存在，因为我还没有弄清楚如何使用 astyanax 创建它，并且该示例没有任何用于创建列族的代码。
package cassandra;
import java.util.Iterator;
import com.netflix.astyanax.ColumnListMutation;
import com.netflix.astyanax.serializers.AsciiSerializer;
import com.netflix.astyanax.serializers.LongSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.netflix.astyanax.AstyanaxContext;
import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.NodeDiscoveryType;
import com.netflix.astyanax.connectionpool.OperationResult;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.connectionpool.impl.ConnectionPoolConfigurationImpl;
import com.netflix.astyanax.connectionpool.impl.CountingConnectionPoolMonitor;
import com.netflix.astyanax.impl.AstyanaxConfigurationImpl;
import com.netflix.astyanax.model.Column;
import com.netflix.astyanax.model.ColumnFamily;
import com.netflix.astyanax.model.ColumnList;
import com.netflix.astyanax.thrift.ThriftFamilyFactory;
public class AstClient {
private static final Logger logger = LoggerFactory.getLogger(AstClient.class);
private AstyanaxContext<Keyspace> context;
private Keyspace keyspace;
private ColumnFamily<Long, String> EMP_CF;
private static final String EMP_CF_NAME = "employees2";
public void init() {
logger.debug("init()");
context = new AstyanaxContext.Builder()
.forCluster("Test Cluster")
.forKeyspace("test1")
.withAstyanaxConfiguration(new AstyanaxConfigurationImpl()
.setDiscoveryType(NodeDiscoveryType.RING_DESCRIBE)
)
.withConnectionPoolConfiguration(new ConnectionPoolConfigurationImpl("MyConnectionPool")
.setPort(9160)
.setMaxConnsPerHost(1)
.setSeeds("127.0.0.1:9160")
)
.withAstyanaxConfiguration(new AstyanaxConfigurationImpl()
.setCqlVersion("3.0.0")
.setTargetCassandraVersion("2.0.5"))
.withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
.buildKeyspace(ThriftFamilyFactory.getInstance());
context.start();
keyspace = context.getClient();
EMP_CF = ColumnFamily.newColumnFamily(
EMP_CF_NAME,
LongSerializer.get(),
AsciiSerializer.get());
}
public void insert(long time) {
MutationBatch m = keyspace.prepareMutationBatch();
ColumnListMutation<String> x =
m.withRow(EMP_CF, time);
for(int i=0;i<10000;i++)
x.putColumn(Integer.toString(i), Integer.toString(i));
try {
@SuppressWarnings("unused")
Object result = m.execute();
} catch (ConnectionException e) {
logger.error("failed to write data to C*", e);
throw new RuntimeException("failed to write data to C*", e);
}
logger.debug("insert ok");
}
public void createCF() {
}
public void read(long time) {
OperationResult<ColumnList<String>> result;
try {
result = keyspace.prepareQuery(EMP_CF)
.getKey(time)
.execute();
ColumnList<String> cols = result.getResult();
// process data
// a) iterate over columsn
for (Iterator<Column<String>> i = cols.iterator(); i.hasNext(); ) {
Column<String> c = i.next();
String v = c.getStringValue();
System.out.println(c.getName() + " " + v);
}
} catch (ConnectionException e) {
logger.error("failed to read from C*", e);
throw new RuntimeException("failed to read from C*", e);
}
}
public static void main(String[] args) {
AstClient c = new AstClient();
c.init();
long t00 = System.nanoTime();
for(int i=0;i<1000;i++) {
long t0 = System.nanoTime();
c.insert(i);
long dt = System.nanoTime() - t0;
System.out.println((1.0e9/dt) + " " + i);
}
long dtt = System.nanoTime() - t00;
c.read(0);
System.out.println(dtt / 1e9);
}
}
更新：我在cassandra-user 邮件列表中找到了这个帖子。在进行大的宽行插入时，CQL 似乎存在性能问题。有一张票 CASSANDRA-6737 来跟踪这个问题。
Update2：我已经试用了 CASSANDRA-6737 附带的补丁，我可以确认这个补丁完全解决了这个问题。感谢 DataStax 的 Sylvain Lebresne 如此迅速地解决了这个问题！
【问题讨论】：
标签： performance cassandra cql3 datastax-java-driver