已解决-Knex js流大数据

Question

我有一个 MySQL table 有数百万条数据。对于每一行，我必须应用自定义逻辑并在另一个 table.

上更新修改后的数据

使用 knex.js 我运行使用 stream() 函数读取数据的查询

获得 Stream 对象后，我将我的逻辑应用于数据事件。一切正常，但在某个时刻它停止而没有给出任何错误。

我尝试在新的 table 中每次更新操作之前暂停流，并在完成更新后重新启动它，但问题没有解决。尝试限制查询，例如限制为 1000 个结果，系统工作正常。

示例代码：

const readableStream = knex.select('*')
  .from('big_table')
  .stream();

readableStream.on('data', async(data) => {
    readableStream.pause() // pause stream
    const toUpdate = applyLogic(data) // sync func
    const whereCond = getWhereCondition(data) // sync func
    try {
       await knex('to_update').where(whereCond).update(toUpdate)
       console.log('UPDATED')
       readableStream.resume() // resume stream
    } catch (e) {
       console.log('ERROR', e)
    }

    readableStream.resume() // resume stream
    
}).on('finish', () => {
   console.log('FINISH')
}).on('error', (err) => {
   console.log('ERROR', err)
})

谢谢！

Answer 1

我解决了

问题不是 knex.js 或流引起的，而是我的开发环境引起的。我在gcp上用k3d模拟生产环境。因此，为了在本地测试我的脚本，我对 MySQL 服务进行了 port-forward。

我不清楚系统崩溃的原因，但通过使用我的脚本创建一个容器以便它连接到 MySQL 服务，该算法按我预期的方式工作。

谢谢

已解决-Knex js流大数据

Solved - Knex js stream large data

mysql

node.js

kubernetes

knex.js