在 NodeJS 中使用转换/双工流答案

【问题标题】：Using transform/duplex stream in NodeJS在 NodeJS 中使用转换/双工流
【发布时间】：2019-03-18 08:55:24
【问题描述】：

我正在使用 NodeJS（noble 模块）从外部来源（蓝牙低功耗）收集数据。我将它们流式传输到外部文件中的 JSON 数组。我们称它为 fromSource.json。它看起来像这样：

来自Source.json

[
     {
         "id": 1,
         "name": "foo"
         "value": 123
     },
     {
         "id": 2,
         "name": "foo2",
         "value": 123
     },
     {
         "id": 3,
         "name": "foo3",
         "value": 789
     }
]

另一方面，我将实时处理这些对象并将新值存储在 CSV 文件中。我们称它为 toDestination.csv。它看起来像这样：

toDestination.csv

id,name,convertedValue
1,"foo",123000
2,"foo2",456000
3,"foo3",789000

每一秒，我都会从源接收一个新值（一个新的 json 对象），将其推送到可写流（json 数组文件）中，然后将其读入可读流，进行转换，然后将其再次写入其最终目的地，即 csv 文件。

我的问题是：NodeJS 流是否适合处理 JSON 对象？我应该在使用它们之前对它们进行字符串化吗？我应该使用 Duplex 还是 Transform 流？

【问题讨论】：

您的源发送什么字节？每个 JSON 对象后是否有换行符？来自您的源的流看起来像 node.js 流对象吗？（管道，on（'数据'等）

标签： javascript node.js json csv

【解决方案1】：

根据您的问题，我认为 Transform 就是您所需要的。在 Duplex 中，您需要同时实现读取和写入，这在您的情况下是不必要的。

代码如下所示：

const intermediate = measurementStream.pipe(
    new Transform({transform: initialTransforms})
);
intermediate.pipe(someArrayifyStreamToJSON)
intermediate.pipe(
    new Transform({transform: someMoreTransforms})
).pipe(
    new Transform({transform: stringifyToCSV})
).pipe(
    fs.createWriteStream('some_path.csv')
);

我还建议查看我创建的框架并支持scramjet。它旨在处理像您这样的情况，并使您的代码更简单：

const {DataStream} = require('scramjet');

// Pass your original stream here, this could be also an 
// AsyncGenerator in node v10 and up.
DataStream.from(measurementStream)
     // you can map with any async or sync operation on the data
    .map(async item => {
         const newData = await doSomeTransforms();
         return newData;
    })
    // tee creates a stream at this point in transforms
    .tee(stream => stream
        .toJSONArray()
        .pipe(fs.createWriteStream('your-intermediate.json'))
    )
    // then you can add some more transforms
    .map(someMapper)
    .filter(someFilter)
    .CSVStringify()
    .pipe(fs.createWriteStream('your-final.csv');

如果你还是选择第一个路径，我会推荐几个模块，让你的生活更轻松：JSONStream 和 papaparse 都在 NPM 中可用。

【讨论】：