如何实现SQL Server到Cassandra数据迁移及大批量插入报错解决

现有代码核心问题 #

• 事件回调与异步逻辑不匹配：tedious的row事件是同步触发的，不会等待回调内的await插入操作完成，短时间内会生成上万并发的Cassandra写入请求，直接压垮集群，触发写入过载报错
• 变量作用域错误：data、dataMap定义在回调外层，并发处理多行数据时会互相覆盖，导致写入数据错乱
• 逐行写入效率极低，没有利用Cassandra的批量写入能力
• 资源关闭逻辑错误：单次插入失败、doneInProc触发时直接关闭客户端，会丢失大量还在队列中的写入请求

优化实现方案 #

核心优化点 #

• 用限流队列控制写入并发，通常Cassandra单节点写入并发控制在100-300是合理区间，避免超过集群处理上限
• 攒批写入，每攒够200-500行执行一次批量插入，大幅降低网络交互开销
• 实现背压机制：当待写入的队列达到阈值时，暂停SQL Server的查询读取，等队列消费到低位再恢复读取，避免内存溢出
• 所有行处理完成后，等待剩余的写入请求全部执行完成再关闭连接

示例代码 #

const { Queue } = require('p-queue');
// 配置限流队列，并发数根据Cassandra集群规模调整，3节点集群建议设为200
const insertQueue = new Queue({ concurrency: 200 });
// 批量写入阈值，可根据单行数据大小调整，建议单批次总大小不超过1MB
const BATCH_SIZE = 300;
let batchBuffer = [];
let totalProcessed = 0;

request.on('row', function (columns) {
    // 变量定义在回调内部，避免并发覆盖
    const data = {};
    const dataMap = {};
    columns.forEach(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;
        }
    });
    data.dataMap = dataMap;
    // 按实际业务补全插入参数顺序
    const insertParams = [data.timestamp, data.partitionName, data.dataMap];
    batchBuffer.push(insertParams);

    // 攒够批次就提交写入
    if (batchBuffer.length >= BATCH_SIZE) {
        const batchToInsert = [...batchBuffer];
        batchBuffer = [];
        // 把写入任务加到限流队列
        insertQueue.add(() => executeBatchInsert(batchToInsert));
    }

    // 队列待处理任务超过阈值，暂停SQL查询读取，实现背压
    if (insertQueue.size > 1000) {
        connection.pause();
    }

    // 队列消费到低位，恢复SQL查询读取
    insertQueue.onEmpty().then(() => {
        if (connection.paused) {
            connection.resume();
        }
    });
});

request.on('doneInProc', async function (rowCount, more) {
    console.log(rowCount + ' rows returned');
    // 把缓冲中剩余的不满一批的数据也提交写入
    if (batchBuffer.length > 0) {
        insertQueue.add(() => executeBatchInsert(batchBuffer));
    }
    // 等待所有写入任务全部完成
    await insertQueue.onIdle();
    console.log(`All ${totalProcessed} rows inserted successfully`);
    // 全部完成再关闭连接
    await cassandraConnection.client.shutdown();
    connection.close();
});

// 批量插入方法
async function executeBatchInsert(paramsList) {
    const insertString = `INSERT INTO ${cassandraKeyspace}.${cassandraTableName}(${cassandraDateColumn}, month_year_index, dataMap) VALUES (?,?,?)`;
    // 生成批量查询语句，不需要跨分区原子性可开启UNLOGGED模式提升性能
    const batchQueries = paramsList.map(params => ({
        query: insertString,
        params: params
    }));
    try {
        await cassandraConnection.client.batch(batchQueries, { prepare: true, logged: false });
        totalProcessed += paramsList.length;
        console.log(`Inserted batch of ${paramsList.length} rows, total processed: ${totalProcessed}`);
    } catch (error) {
        logger.error('Batch insert failed:', error);
        // 可在此补充失败重试逻辑，比如把失败的批次重新加入队列
    }
}

connection.execSql(request);

额外调优建议 #

• 如果单行数据很大，适当降低BATCH_SIZE，避免单批次过大触发Cassandra的批量大小限制
• 可调整限流队列的并发数，根据集群负载情况逐步上调，直到CPU/IO使用率达到合理区间
• 迁移前可以把Cassandra表的一致性级别调低到LOCAL_ONE，迁移完成后再改回业务需要的级别，进一步提升写入速度
• 避免在分区键分布不均匀的场景下使用过大的批量，防止单节点压力过高

内容的提问来源于stack exchange，提问作者XXSD