使用 node.js 将数据流保存到 Cassandra答案

【问题标题】：Saving data stream into Cassandra using node.js使用 node.js 将数据流保存到 Cassandra
【发布时间】：2018-05-26 06:07:22
【问题描述】：

我有一个数据流（通过 Node EventEmitter）以 JSON 格式发射数据，并希望在发射时将流保存到 Cassandra。有没有一种优雅的方式来实现这个功能？

我使用的驱动是 nodejs-dse-driver，Cassandra 版本是 3.11.1。请建议是否有任何推荐的插件可以用来完成上述任务。

【问题讨论】：

标签： node.js cassandra datastax-enterprise cassandra-3.0

【解决方案1】：

您可以从源中分块读取数据并并行发送，例如（使用async library）：

const limit = 10;
stream.on('readable', () => {
  let r;
  const rows = [];
  async.whilst(function condition() {
    while ((r = csv.read()) != null && rows.length < limit) {
      rows.push(r);
    }
    return rows.length > 0;
  }, function eachGroup(next) {
    // we have a group of 10 rows or less to save
    // we can do it in a batch
    // or we can do it in parallel with async.each()
    async.each(rows, (r, eachCallback) {
      // Adapt the row to parameters
      // For example: sample
      const params = r.split(',);
      client.execute(query, params, { prepare: true}, eachCallback);
    }, next);
  }, function groupFinished(err) {
    if (err) {
      // something happened when saving
      // TODO: do something with err
      return;
    }
    // This chunk of rows emitted by stream where saved
  });
}).on('end', () => {
  // no more data from source
});

【讨论】：

感谢您提供上述示例。我现在没有使用异步，但在读取流 (.on('data')) 函数中，我通过 client.execute 函数执行 Cassandra 插入。此设置适用于较小的数据集，而如果数据集很大，我最终会出现以下错误，我尝试插入 5000+ 行并弹出超时错误，错误：所有主机尝试查询失败。第一个主机尝试，:9042: OperationTimedOutError: 主机 :9042 在超时 12000 毫秒之前没有回复。请参阅内部错误。
当执行一个简单的查询来获取发生插入操作的表的计数时，我最终得到了同样的错误。我拥有的 Cassandra 集群非常小，只有 1 个副本，这是因为集群大小而发生的吗？

【解决方案2】：

这是转换流的一个很好的用例。

如果您有一个真正的可读流，那么您可以将任何转换流通过管道传输到可读流中。我不认为事件发射器是可读流，因此您可能需要更改原始数据获取实现。

有关实现细节，请参阅 NodeJS 文档。 https://nodejs.org/api/stream.html#stream_new_stream_transform_options

这取决于您的 NodeJS 版本。

const myTransformStream = new Transform({
  objectMode: true,
  transform(row, encoding, callback) {
    // insert into Cassandra code here
    cassandra.execute(query, row, {prepare: true}, () => {
        // after the execute is done, callback to process more 
        callback(null, row);
    });
  }
});

originalStream.pipe(myTransformStream);

【讨论】：