如何使用 node.js 顺序读取 csv 文件（使用流 API）

Posted 2023-03-28

【中文标题】如何使用 node.js 顺序读取 csv 文件（使用流 API）

【英文标题】：How to sequentially read a csv file with node.js (using stream API)

【发布时间】：2022-01-15 05:58:49

【问题描述】：

我试图弄清楚如何创建一个流管道，它可以按需读取 csv 文件中的条目。为此，我想到了使用管道（伪代码）使用以下方法

const stream_pipe = input_file_stream.pipe(csv_parser)
// Then getting entries through:
let entry = stream_pipe.read()

不幸的是，经过大量测试后，我发现它们在我设置管道的那一刻，它会自动消耗，直到 csv 文件结束。我试图通过在末尾附加.pause() 来暂停它，但它似乎没有任何效果。

这是我当前的代码。我正在使用csv_parse 库（更大的csv 包的一部分）：

// Read file stream
const file_stream = fs.createReadStream("filename.csv")
const parser = csvParser(
    columns: ['a', 'b'],
    on_record: (record) => 
        // A simple filter as I am interested only in numeric entries
        let a = parseInt(record.a)
        let b = parseInt(record.b)
        return (isNaN(a) || isNaN(b)) ? undefined : record
    
)
const reader = stream.pipe(parser) // Adding .pause() seems to have no effect
console.log(reader.read()) // Prints `null`

// I found out I can use this strategy to read a few entries immediately, but I cannot break out of it and then resume as the stream will automatically be consumed 
//for await (const record of reader) 
//    console.log(record)
// 

我一直在努力解决这个问题，但在 csv 包和节点官方文档上都找不到简单的解决方案。

提前感谢任何能让我走上正轨的人:)

【参考方案1】：

您可以在读取流时做一件事，您可以创建一个 readLineInterface 并像这样传递输入流和正常输出流：

const inputStream = "reading the csv file",
      outputStream = new stream();

// now create a readLineInterface which will read 
// line by line you should use async/await 

const res = await processRecord(readline.createInterface(inputStream, outputStream));

async function processRecord(line) 
   return new Promise((res, rej) => 
       if (line) 
        // do the processing 
        res(line);
       
       
       rej('Unable to process record');
   )

现在创建processRecord 函数应该逐行获取内容，并且您可以承诺使其按顺序排列。

注意：上面的代码是一个伪代码，只是为了让你知道事情是否有效，因为我在我的项目中一直在做同样的事情来读取csv文件的行和行，它工作正常。