使用 Node.js 的流写入和读取大型数组

Question

我有一个巨大的对象，它用作具有 270 万个键的地图。我尝试将对象写入文件系统，以便将其持久化，而不是在每次需要时重新计算它。在另一个步骤中，我需要再次读取对象。我需要访问内存中的整个对象，因为它需要用作地图。
为了写入，我将对象转换为数组并使用以下函数将其流式传输到文件系统。我首先将其转换为数组的原因是流式传输数组而不是对象似乎要快得多。写作部分大约需要一分钟，这很好。输出文件的大小为 4,8GB。
我面临的问题是尝试读取文件时。为此，我创建了一个读取流并解析内容。 但是，由于某种原因，我似乎遇到了某种内存限制。我使用了各种不同的读取和解析方法，它们似乎都可以正常工作，直到读取了大约 50% 的数据（此时我机器上的节点进程占用 6GB 内存，略低于我设置的限制）。从那时起，读取时间显着增加了 10 倍，可能是因为节点接近使用最大分配内存限制（6144MB）。感觉就像我做错了什么。
我不明白的主要事情是为什么写入不是问题，而读取是，即使在写入步骤期间，整个数组也保存在内存中。我正在使用节点v8.11.3。

总结一下：

• 我有一个大对象需要使用流作为数组持久保存到文件系统中
• 写得很好
• 读取工作直到读取了大约 50% 的数据，然后读取时间显着增加

如何才能更高效地读取文件？

我尝试了各种库，例如stream-to-array，read-json-stream， JSONStream

要写入的对象示例：

{ 'id': ['some_other_id_1', 'some_other_id_2'] }

然后在写入之前将其转换为数组：

[{ 'id': ['some_other_id_1', 'some_other_id_2'] }]

使用流将数组写入文件系统的函数：

import * as fs from 'fs'
import * as jsonStream from 'JSONStream'
import * as streamifyArray from 'stream-array'

async function writeFileAsStreamFromArray(pathToFile: string, fileContent: any[]): Promise<void> {
  return new Promise((resolve, reject) => {
    const fileWriterStream = fs.createWriteStream(pathToFile)
    const stringifierStream = jsonStream.stringify()
    const readStream = streamifyArray(fileContent)
    readStream.pipe(stringifierStream)
    stringifierStream.pipe(fileWriterStream)

    fileWriterStream.on('finish', () => {
      console.log('writeFileAsStreamFromArray: File written.')
      stringifierStream.end()
      resolve()
    })
    fileWriterStream.on('error', (err) => {
      console.log('err', err)
      reject(err)
    })
  })
}

使用 jsonStream 从流中获取数组的函数：

async function getArrayFromStreamUsingJsonStream(pathToFile: string): Promise<any[]> {
  return new Promise(async (resolve, reject) => {
    const readStream = fs.createReadStream(pathToFile)
    const parseStream = jsonStream.parse('*')
    const array = []
    const start = Date.now()

    const transformer = transform((entry) => {
      array.push(entry)
      if ((array.length % 100000) === 0) {
        const end = (Date.now() - start) / 1000
        console.log('array', array.length, end)
      }
    })
    readStream.pipe(parseStream)
    parseStream.pipe(transformer)

    readStream.on('end', () => {
      console.log('getArrayFromStreamUsingJsonStream: array created')
      parseStream.end()
      resolve(array)
    })
    readStream.on('error', (error) => {
      reject(error)
    })
  })
}

计时日志（在 1200000 条记录后，我取消了执行，因为它花了很长时间）：

array 100000 6.345
array 200000 12.863
array 300000 21.177
array 400000 29.638
array 500000 35.884
array 600000 42.079
array 700000 48.74
array 800000 65.662
array 900000 89.805
array 1000000 120.416
array 1100000 148.892
array 1200000 181.921
...

预期结果：应该比目前的性能更高。 这甚至可能吗？还是我遗漏了一些明显的东西？

非常感谢任何帮助！

Answer 1

我怀疑它内存不足，因为您试图将所有条目读入一个连续数组。随着数组填满，节点将重新分配数组并将现有数据复制到新数组。所以随着数组变得越来越大，它变得越来越慢。因为在重新分配时它必须有两个数组，所以它也会使用比数组本身更多的内存。

您可以使用数据库，因为几百万行应该不是问题，或者编写您自己的读/写例程以确保您使用允许非顺序块内存分配的东西，例如https://www.npmjs.com/package/big-array

例如，预分配一个 10k 条目长的数组，将映射的前 10k 条目读入数组并将数组写入文件。然后将接下来的 10k 个条目读入数组并将其写入一个新文件。重复直到处理完所有条目。这应该会减少您的内存使用量，并通过以使用更多内存为代价并行运行 IO 来加快速度。