我正在评估Cassandra。我正在使用DataStax驱动程序和CQL。
我想要存储一些具有以下内部结构的数据，其中每次更新的名称都不同。
我希望模式是这样的，以便于查询。我偶尔需要存储超过65000行的更新。因此，使用cassandra列表/集合/映射数据类型不是一个选项。
我必须能够处理每秒至少1000个宽行插入，其中包含数量不定但很大（~1000）的名称/值对。
问题如下：我编写了一个简单的基准测试，每次进行1000个宽行插入，每个宽行包含10000个名称/值对。使用CQL和datastax驱动程序时，性能非常慢，而不使用CQL（使用astyanax）的版本在同一测试集群上具有良好的性能。
我已阅读related question，在该问题的被接受的答案中建议您应该能够通过使用批处理预准备语句（在cassandra 2中可用）原子且快速地创建新的宽行。
所以我尝试使用它们，但我仍然获得缓慢的性能（对于在本地主机上运行的小型三节点集群每秒两个插入）。我是否漏掉了一些显而易见的东西，还是必须使用较低级别的thrift API？我已经在astyanax中使用ColumnListMutation实现了相同的插入，并且每秒获得约30个插入。
如果我必须使用较低级别的thrift API：
- 它是否实际上已被弃用，还是因为它是较低级别而不便于使用？ - 我能够使用CQL查询使用thrift api创建的表吗？
下面是一个包含在scala中的自包含代码示例。它只是创建了一个批处理语句，用于插入具有10000列的宽行，并重复计时插入性能。
我尝试了BatchStatement和一致性级别的选项，但没有任何方法可以让我获得更好的性能。
唯一的解释是，尽管批处理由准备好的语句组成，但条目仍逐一添加到行中。
更新：我在cassandra-user邮件列表中找到了这个主题。当进行大型宽行插入时，使用CQL存在性能问题。有一个票CASSANDRA-6737来跟踪此问题。
更新2：我已经尝试了附加到CASSANDRA-6737的补丁，并确认该补丁完全修复了此问题。感谢DataStax的Sylvain Lebresne如此快速地解决了这个问题！
我想要存储一些具有以下内部结构的数据，其中每次更新的名称都不同。
+-------+-------+-------+-------+-------+-------+
| | 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列表/集合/映射数据类型不是一个选项。
我必须能够处理每秒至少1000个宽行插入，其中包含数量不定但很大（~1000）的名称/值对。
问题如下：我编写了一个简单的基准测试，每次进行1000个宽行插入，每个宽行包含10000个名称/值对。使用CQL和datastax驱动程序时，性能非常慢，而不使用CQL（使用astyanax）的版本在同一测试集群上具有良好的性能。
我已阅读related question，在该问题的被接受的答案中建议您应该能够通过使用批处理预准备语句（在cassandra 2中可用）原子且快速地创建新的宽行。
所以我尝试使用它们，但我仍然获得缓慢的性能（对于在本地主机上运行的小型三节点集群每秒两个插入）。我是否漏掉了一些显而易见的东西，还是必须使用较低级别的thrift 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来跟踪此问题。
更新2：我已经尝试了附加到CASSANDRA-6737的补丁，并确认该补丁完全修复了此问题。感谢DataStax的Sylvain Lebresne如此快速地解决了这个问题！