使用c#在cassandra中批量插入

Question

我是卡桑德拉的新人。我必须使用c#在卡桑德拉中一次性插入50000行。我正在使用卡桑德拉c#驱动程序。我正在使用以下代码在卡桑德拉中插入数据。请帮帮我

string tableName = "" + ConfigurationManager.AppSettings["tableName"];
            string keySpace = "test";// "" + ConfigurationManager.AppSettings["Keyspace"];
            //string query = "INSERT INTO " + tableName + " (emp_name, emp_position,emp_firstname,uniqueid) VALUES (?, ?,?,?)"; //; "SELECT * FROM "+ tableName + " limit 2 ";
            string query = string.Empty;
            query =
"BEGIN BATCH " +
"INSERT INTO " + tableName + " (emp_name, emp_position,emp_firstname,uniqueid) VALUES (?, ?,?,?);" +
"INSERT INTO " + tableName + " (emp_name, emp_position,emp_firstname,uniqueid) VALUES (?, ?,?,?);" +
"INSERT INTO " + tableName + " (emp_name, emp_position,emp_firstname,uniqueid) VALUES (?, ?,?,?);" +
" APPLY BATCH";

BatchStatement(keySpace, query,
                    Convert.ToString("ashish" + i), 2, Convert.ToString("Mohan" + i), System.Guid.NewGuid(),
                    Convert.ToString("ashish" + i), 2, Convert.ToString("Mohan" + i), System.Guid.NewGuid(),
                    Convert.ToString("ashish" + i), 2, Convert.ToString("Mohan" + i), System.Guid.NewGuid()

public void BatchStatement(string keySpace, string query, params object[] parameter)
    {
        try
        {
            BatchStatement objBatchStatement = new BatchStatement();
            PreparedStatement statement = PrepareQueryStatement(keySpace, query);
            objBatchStatement.Add(statement.Bind(parameter));
            var session = cluster.Connect(keySpace);
            //objBatchStatement.ConsistencyLevel
            // Execute the batch
            //RowSet row = session.Execute(query);
            RowSet row = session.Execute(objBatchStatement);

        }
        catch (Exception ex)
            {
            Console.WriteLine("Excpetion occured during batch operation method Name BatchStatement error : " + ex.ToString());
        }

    }

我收到错误：批处理中的语句无效：只允许UPDATE、INSERT和DELETE语句。

Answer 1

要回答具体问题...

批处理中的语句无效：仅允许更新、插入和删除语句。

"BEGIN BATCH " +
...
" APPLY BATCH";

< code>BEGIN BATCH和< code>APPLY BATCH是不允许的语句。< code>BatchStatement对象会为您处理这些问题。

至于这种说法:

我得在Cassandra一气呵成插入50000行

João是绝对正确的，因为您正在做的是滥用Cassandra的BATCH功能。BATCH（在Cassandra中）旨在原子地将一次写入应用于多个表。它不是为支持50000次写入一个表而构建的。我见过开发团队在这样做时无意中导致集群节点崩溃。

这不起作用的原因是，卡桑德拉意识到它没有一个单独的分区来发送批处理。因此，它选择一个协调器节点，该节点负责从所有其他节点提取和组合50000个结果。该节点很快就会不堪重负，并崩溃。

同样，João 关于异步编写行的建议是要走的路。我要做的唯一调整是添加一种机制来限制任何时候处于活动状态的线程数，作为保护您的节点免受50k次背压写入的一种方式。

Answer 2

我不太理解您的代码示例，所以我不知道您是否在批处理中使用了< code>SELECT查询，这似乎是问题所在。无论如何，您可能会误用< code>Batch,因为Cassandra中的批处理用于原子性而不是性能(大多数情况下)。

如果您想尽可能快地插入这50000行，那么使用session.ExecuteAsync异步执行50000个插入会更快（并且建议）。尝试这样做：

var tasks = new List<Task>();

foreach (BoundStatement query in queries) 
{
    tasks.Add(session.ExecuteAsync(query));
}

await Task.WhenAll(tasks).ConfigureAwait(false);

对于需要插入更多行的更高级的用例，可能需要控制并发执行的请求数量，但标准任务并行库已经为您完成了其中的一些工作。

批处理比单独异步执行所有这些请求更快的唯一情况是，您可以使用属于同一分区的语句创建微批处理，但这要困难得多，并且很可能不需要这种性能提升。

在您编辑了您的问题后，我可以理解代码示例。如果您仍然想使用Batch而不是我上面建议的方法，那么您要做的是：

var batch = new BatchStatement();
foreach (BoundStatement query in queries) 
{
    batch.Add(query); // you can also add SimpleStatements instead of BoundStatements
}

await session.ExecuteAsync(batch).ConfigureAwait(false);