使用 datastax cpp 驱动程序改进 cassandra 数据库中的插入时间答案

【问题标题】：improve insertion time in cassandra database with datastax cpp driver使用 datastax cpp 驱动程序改进 cassandra 数据库中的插入时间
【发布时间】：2016-09-13 15:28:40
【问题描述】：

我有一个使用 datastax cpp 驱动程序填充的简单表。该表存储在 cassandra 数据库中。该表是这样创建的：

create table data (
    dt_id int PRIMARY KEY,
    dt_numbers list<int>,
    insertion_time timestamp,
)

所以，对于我插入的每一行

VALUES (1, [1,2,3,4], now())

问题是我尝试插入100万行σ，时间是35分钟。在生产环境中，该时间被认为很长。

一些伪代码：

std::map<int, vector<int>> myData;
for( all entries in myData) {
  const char* query = "INSERT INTO...";
  future = cass_session_execute(session, statement);
  cass_future_wait(future);
}

数据存储在 std::map 中，我不关心首先写入哪个地图索引，但每个地图条目都应该准确插入一个。有什么方法可以提高这个程序的性能吗？有例子吗？

到目前为止，我使用的代码是一行接一行地编写，与其中一个 datastax 示例中使用的代码相似

    CassError insert_into_basic(CassSession* session, const char* key, const Basic* basic) {
      CassError rc = CASS_OK;
      CassStatement* statement = NULL;
      CassFuture* future = NULL;
      const char* query = "INSERT INTO examples.basic (key, bln, flt, dbl, i32, i64) VALUES (?, ?, ?, ?, ?, ?);";

      statement = cass_statement_new(query, 6);

      cass_statement_bind_string(statement, 0, key);
      cass_statement_bind_bool(statement, 1, basic->bln);
      cass_statement_bind_float(statement, 2, basic->flt);
      cass_statement_bind_double(statement, 3, basic->dbl);
      cass_statement_bind_int32(statement, 4, basic->i32);
      cass_statement_bind_int64(statement, 5, basic->i64);

      future = cass_session_execute(session, statement);
      cass_future_wait(future);

      rc = cass_future_error_code(future);
      if (rc != CASS_OK) {
        print_error(future);
      }

      cass_future_free(future);
      cass_statement_free(statement);

      return rc;
    }

【问题讨论】：

标签： c++ cassandra datastax

【解决方案1】：

您的性能变慢了，因为您一次写入一行并等待该单行被真正写入。如果您仔细查看文档，您会发现要从 C* 中获得最佳性能，您确实需要执行异步写入。在伪代码中它会是这样的：

std::map<int, vector<int>> myData;
std::list<CassFuture> futures;
for( all entries in myData) {
  const char* query = "INSERT INTO...";
  futures.push_back(cass_session_execute(session, statement));
  if (futures.size() > 5000) {
    for (all entries in futures) {
       cass_future_wait(future);
       cass_future_free(future);
    }
    futures.clear();
  }
}
// Wait for the "trailing" futures... 
for (all entries in futures) {
   cass_future_wait(future);
   cass_future_free(future);
}

这样您应该会获得性能提升。您应该需要调整 5000 以匹配您的要求/硬件规格。当然，还要实施一些策略来处理写入失败。

【讨论】：