如何处理大文件、NodeJS 流和管道

Question

我对 NodeJS 流有点陌生，我越了解它，我就越相信它不是 特别简单稳定的东西。我正在尝试读取大文件 csv / csv-parse（显然是最流行的 CSV 模块 使用 NodeJS）使用piping API，它 涉及使用同一作者的stream-transform。

我在这里体验到的部分内容实际上可以在不实际使用解析器的情况下重现，所以 我对这些部分进行了注释以使示例更简单（对于那些更喜欢 JavaScript 而不是 CoffeeScript 的人， there's also a JS version):

#-------------------------------------------------------------------------------
fs                        = require 'fs'
transform_stream          = require 'stream-transform'
log                       = console.log
as_transformer            = ( method ) -> transform_stream method, parallel: 11
# _new_csv_parser           = require 'csv-parse'
# new_csv_parser            = -> _new_csv_parser delimiter: ','

#-------------------------------------------------------------------------------
$count = ( input_stream, title ) ->
  count = 0
  #.............................................................................
  input_stream.on 'end', ->
    log ( title ? 'Count' ) + ':', count
  #.............................................................................
  return as_transformer ( record, handler ) =>
    count += 1
    handler null, record

#-------------------------------------------------------------------------------
read_trips = ( route, handler ) ->
  # parser      = new_csv_parser()
  input       = fs.createReadStream route
  #.............................................................................
  input.on 'end', ->
    log 'ok: trips'
    return handler null
  input.setMaxListeners 100 # <<<<<<
  #.............................................................................
  # input.pipe parser
  input.pipe $count input, 'trips A'
    .pipe $count    input, 'trips B'
    .pipe $count    input, 'trips C'
    .pipe $count    input, 'trips D'
    # ... and so on ...
    .pipe $count    input, 'trips Z'
  #.............................................................................
  return null

route = '/Volumes/Storage/cnd/node_modules/timetable-data/germany-berlin-2014/trips.txt'
read_trips route, ( error ) ->
  throw error if error?
  log 'ok'

输入文件包含204865行GTFS数据； 我不是在这里解析它，只是直接阅读它，所以我猜我用上面的代码计算的是块 数据。

我正在将流从一个柜台输送到另一个柜台，并且希望尽可能频繁地击中最后一个柜台 第一个;然而，这就是我得到的：

trips A: 157
trips B: 157
trips C: 157
...
trips U: 157
trips V: 144
trips W: 112
trips X: 80
trips Y: 48
trips Z: 16

在我实际解析数据的早期设置中，我得到了这个：

trips A: 204865
trips B: 204865
trips C: 204865
...
trips T: 204865
trips U: 180224
trips V: 147456
trips W: 114688
trips X: 81920
trips Y: 49152
trips Z: 16384

所以看起来溪流在途中会干涸。

我的怀疑是输入流的end 事件不是一个可靠的信号，当 试图确定所有处理是否已完成——毕竟，假设处理是合乎逻辑的 只能在流完全消耗完一段时间后才能完成。

所以我寻找另一个事件来收听（没有找到）并延迟调用回调（使用 setTimeout、process.nextTick 和 setImmediate)，但无济于事。

如果有人能指出来就好了

• (1)setTimeout、process.nextTick 和 setImmediate 在这种情况下的关键区别是什么，以及
• (2)如何可靠地判断最后一个字节是否已经被管道的最后一个成员处理了。

更新我现在认为问题出在流转换上，该问题存在一个问题，有人报告了一个非常相似的问题，几乎相同的数字（他有 234841 条记录，最后是 16390，我有204865 并以 16384 结束）。不是证明，而是太接近了，不会是偶然的。

我放弃了流转换并改用event-stream.map；然后测试运行正常。

Answer 1

几天后我想我可以说stream-transform 对大文件有问题。

我已经切换到event-stream，恕我直言，这是一个更好的整体解决方案，因为它是完全通用的（即它一般是关于流的，而不是特别关于 CSV-data-as-streams）。我已经在我的初始 pipdreams 模块的文档中概述了关于 NodeJS 中的流库的一些想法，该模块提供了许多常用的流操作。