Node.js - 读取 CSV 文件不适用于行号 > 500

Question

我目前正在努力运行我的 Node.js 服务器。

我想做什么：

• 将 CSV 文件从移动设备上传到我的本地服务器并保存在文件系统中
• 读取 .csv 文件的每一行并将每一行保存到我的 MongoDB 数据库中

上传和保存文件完美无缺。读取 .csv 文件并将每一行保存到数据库仅适用于行号较小的文件。 我不知道它停止工作时的确切行数。每次读取文件时似乎都不同。 有时（如果行号大于 1000）我使用的 CSV-Reader 甚至不会开始处理文件。其他时候他只读了 100-200 行然后就停下来了。

这是我如何上传文件的代码：

var fs = require('fs');
var sys = require("sys");
var url = require('url');
var http = require('http');

http.createServer(function(request, response) {
    sys.puts("Got new file to upload!");

    var urlString = url.parse(request.url).pathname;

    var pathParts = urlString.split("/");

    var deviceID = pathParts[1];
    var fileName = pathParts[2];

    sys.puts("DeviceID: " + deviceID);
    sys.puts("Filename: " + fileName);

    sys.puts("Start saving file");
    var tempFile = fs.createWriteStream(fileName);
    request.pipe(tempFile);
    sys.puts("File saved");

    // Starting a new child process which reads the file 
    // and inserts each row to the database
    var task = require('child_process').fork('databaseInsert.js');
    task.on('message', function(childResponse) {
        sys.puts('Finished child process!');
    });
    task.send({
        start : true,
        deviceID : deviceID,
        fileName : fileName
    });
    sys.puts("After task");

    response.writeHead(200, {
        "Content-Type" : "text/plain"
    });
    response.end('MESSAGE');
}).listen(8080);

这一切正常。 现在子进程的代码（databaseInsert.js）：

var sys = require("sys");
var yaCSV = require('ya-csv');
var Db = require('mongodb').Db;
var dbServer = require('mongodb').Server;

process.on('message', function(info) {
    sys.puts("Doing work in child process");

    var fileName = info.fileName;
    var deviceID = info.deviceID;

    sys.puts("Starting db insert!");
    var dbClient = new Db('test', new dbServer("127.0.0.1", 27017, {}), {
        w : 1
    });

    dbClient.open(function(err, client) {
        if (err) {
            sys.puts(err);
        }
        dbClient.createCollection(deviceID, function(err, collection) {
            if (err) {
                sys.puts("Error creating collection: " + err);
            } else {
                sys.puts("Created collection: " + deviceID);

                var csvReader = yaCSV.createCsvFileReader(fileName, {
                    columnsFromHeader : true,
                    'separator' : ';'
                });
                csvReader.setColumnNames([ 'LineCounter', 'Time',  'Activity',
                        'Latitude', 'Longitude' ]);

                var lines = 0;
                csvReader.addListener('data', function(data) {
                    lines++;
                    sys.puts("Line: " + data.LineCounter);
                    var docRecord = {
                        fileName : fileName,
                        lineCounter : data.LineCounter,
                        time : data.Time,
                        activity : data.Activity,
                        latitude : data.Latitude,
                        longitude : data.Longitude
                    };
                    collection.insert(docRecord, {
                        safe : true
                    }, function(err, res) {
                        if (err) {
                            sys.puts(err);
                        }
                    });
                });
            }
        });
    });
    process.send('finished');
});

起初我没有使用子进程，但我的行为与现在相同。所以我对此进行了测试。

希望对 Node.js 有一定经验的人可以帮助我。

Answer 1

我认为您的问题是您正在尝试读取tempFile，而它仍在被写入。现在，您正在将请求传送到文件流（并行和异步进行）并启动读取器进程。然后，读取器进程将开始与写入操作并行读取文件。如果读取器速度更快（通常会更快），它将读取前几条记录，但随后遇到文件结尾并停止读取。

要解决这个问题，您只能在写入完全完成后启动读取器进程，即将sys.puts("File.send"); 之前的部分放入tempFile.end(...) 的回调中（参见http://nodejs.org/api/stream.html#stream_writable_end_chunk_encoding_callback）。

在文件仍在被写入时读取文件，类似于 Unix 中的 tail 命令，在我看来是相当困难的（谷歌了解实现正确尾部有多难的详细信息）。

Answer 2

我没有足够的代表来发表评论（所以不要给我投反对票:)）

但是你熟悉mongoimport/export?

我过去使用它从我的 db 导出到 csv 文件...所以你可以在将它从移动客户端上传到服务器后执行相反的操作。

它来自 shell，但您可以使用 nodeJS_ChildSpawn 编写代码