如何将数据从 SQL Server 迁移到 Cassandra答案

【问题标题】：How to migrate data from SQL Server to Cassandra如何将数据从 SQL Server 迁移到 Cassandra
【发布时间】：2021-08-21 22:33:54
【问题描述】：

我有大量数据要从 SQL Server 传输到 Cassandra。我已经开发了一些代码来使从 SQL Server 接收到的数据与 cassandra 兼容，但我无法插入它。由于它是大量数据，所以我会在一段时间后收到此错误：

我正在使用 nodejs 执行此操作。这是我的代码：

request.on('row', async function (columns) {
    columns.forEach(async function (column) {
        if (column.metadata.colName.toLowerCase() == sqlTimestampCol.toLowerCase()) {
            data.timestamp = column.value;
            data.partitionName = partition_utils.getPartitionNameByDate(column.value);
        } else {
            dataMap[column.metadata.colName] = column.value;
        }
        if (column.value === null) {
            console.log('NULL');
        } else {
            result += column.value + " ";
        }
    });
    data.dataMap = dataMap;

    // INSERT CASSANDRA
    var insertString =
        `INSERT INTO ${cassandraKeyspace}.${cassandraTableName}(${cassandraDateColumn}, month_year_index, dataMap) VALUES (?,?,?)`;

    await cassandraConnection.client.execute(insertString, insertParams, { prepare: true })
        .then(async function () {
            console.log(`Inserted data for ${cassandraTableName} successfully`);
        })
        .catch(async function (error) {
            cassandraConnection.client.shutdown();
            logger.error(error);
        });

    result = "";
    data = {}
});

request.on('doneInProc', function (rowCount, more) {
    console.log(rowCount + ' rows returned');
    cassandraConnection.client.shutdown();
});
connection.execSql(request);

我正在使用繁琐的 npm 和 datastax cassandra 驱动程序。我做了一个简单的select * from table 并为表格的每一行运行上面的代码（在 request.on('row') 中），但这对我不起作用。有没有合适的方法来做到这一点？

从我读到的错误中，我收到的错误是因为我插入数据的速度比 cassandra 处理它的速度快。另外，只有当 SQL 查询结束时，新数据才真正插入到 cassandra 中。

【问题讨论】：

标签： node.js cassandra nosql database-migration

【解决方案1】：

您的代码的问题是您有一个客户端在同一连接上发送数千个查询。与其说是 Cassandra 无法处理请求，不如说是您的客户端和集群之间的单一连接是一个瓶颈。

在您的设置中，每个连接只能处理 2048 个并发请求。达到此限制后，您的应用将无法再发送任何请求。

为了获得最大吞吐量，您需要水平扩展，以便让多个应用实例都发送查询。例如，如果您有 6 个应用实例，那么您可以向 Cassandra 集群发送 6 x 2048 个并发请求。

附带说明，如果这是一次性将您的数据迁移到 Cassandra，那么我建议您将数据导出为 CSV 格式并使用DataStax Bulk Loader (DSBulk)。它是一个完全开源的免费实用程序，允许您在 Apache Cassandra 集群、DSE 集群或Astra 数据库中加载/卸载数据。

有关 DSBulk 的详细信息，请参阅以下内容：

【讨论】：

【解决方案2】：

限制进行中请求的数量需要手动处理流背压。

Node.js 驱动程序为此提供了实用方法：

https://docs.datastax.com/en/developer/nodejs-driver/4.6/features/concurrent-api/#using-a-fixed-query-and-a-readable-stream.

例如：

const stream = myRowStream.pipe(transformRowToArrayOfParameters);
const result = await executeConcurrent(client, insertString, stream);

【讨论】：