在 nodejs 中写入文件之前对数据流进行排序

Question

我有一个可能包含多达 1M 条记录的输入文件，每条记录看起来像这样

field 1 field 2 field3 \n

我想读取这个输入文件并根据field3 对其进行排序，然后再将其写入另一个文件。

这是我目前所拥有的

var fs = require('fs'),
    readline = require('readline'),
    stream = require('stream');

var start = Date.now();

var outstream = new stream;
outstream.readable = true;
outstream.writable = true;

var rl = readline.createInterface({
    input: fs.createReadStream('cross.txt'),
    output: outstream,
    terminal: false
});

rl.on('line', function(line) {
    //var tmp = line.split("\t").reverse().join('\t') + '\n';
    //fs.appendFileSync("op_rev.txt", tmp );
    // this logic to reverse and then sort is too slow
});

rl.on('close', function() {
    var closetime = Date.now();
    console.log('Read entirefile. ', (closetime - start)/1000, ' secs');
});

我基本上被困在这一点上，我所拥有的只是从一个文件读取并写入另一个文件的能力，有没有办法在写入之前有效地对这些数据进行排序

Answer 1

我遇到了非常相似的问题，需要执行external sort。

我发现，在浪费了一些时间之后，我可以将数据加载到数据库中，然后从中查询出所需的数据。

插入没有排序也没关系，只要我的查询结果可以。

希望它也对你有用。

为了将数据插入数据库，节点上有很多工具可以执行此类任务。我有 this pet project 做类似的工作。

我也确信，如果您 search 主题，您会找到更多信息。

祝你好运。

Answer 2

您有两种选择，具体取决于正在处理的数据量。 （3列的1M记录数并不能说明实际数据量）

将数据加载到内存中，就地排序

var lines = [];
rl.on('line', function(line) {
    lines.push(line.split("\t").reverse());
});

rl.on('close', function() {
    lines.sort(function(a, b) { return compare(a[0], b[0]); });

    // write however you want
    fs.writeFileSync(
        fileName,
        lines.map(function(x) { return x.join("\t"); }).join("\n")
    );
    function compare(a, b) {
        if (a < b) return -1;
        if (a > b) return 1;
        return 0;
    }
});

将数据加载到持久化数据库中，按顺序读取

使用您选择的数据库引擎（例如 nedb，一个用于 nodejs 的纯 javascript 数据库）

编辑：似乎 NeDB 将整个数据库保存在内存中，文件只是数据的持久副本。我们将不得不寻找另一个实现。 TingoDB 看起来很有希望。

// This code is only to give an idea, not tested in any way

var Datastore = require('nedb');
var db = new Datastore({
    filename: 'path/to/temp/datafile',
    autoload: true
});

rl.on('line', function(line) {
    var tmp = line.split("\t").reverse();
    db.insert({
        field0: tmp[0],
        field1: tmp[1],
        field2: tmp[2]
    });
});

rl.on('close', function() {
    var cursor = db.find({})
            .sort({ field0: 1 }); // sort by field0, ascending
    var PAGE_SIZE = 1000;
    paginate(0);
    function paginate(i) {
        cursor.skip(i).take(PAGE_SIZE).exec(function(err, docs) {
            // handle errors

            var tmp = docs.map(function(o) {
                return o.field0 + "\t" + o.field1 + "\t" + o.field2 + "\n";
            });
            fs.appendFileSync("op_rev.txt", tmp.join(""));
            if (docs.length >= PAGE_SIZE) {
                paginate(i + PAGE_SIZE);
            } else {
                // cleanup temp database
            }
        });
    }
});

Answer 3

您可以利用流进行此类操作。有一些 NPM 模块会很有帮助——首先通过运行来包含它们

npm install sort-stream csv-parse stream-transform

从命令行。

然后：

var fs = require('fs');
var sort = require('sort-stream');
var parse = require('csv-parse');
var transform = require('stream-transform');

// Create a readble stream from the input file.
fs.createReadStream('./cross.txt')
  // Use `csv-parse` to parse the input using a tab character (\t) as the 
  // delimiter. This produces a record for each row which is an array of 
  // field values.
  .pipe(parse({
    delimiter: '\t'
  }))
  // Use `sort-stream` to sort the parsed records on the third field. 
  .pipe(sort(function (a, b) {
    return a[2].localeCompare(b[2]);
  }))
  // Use `stream-transform` to transform each record (an array of fields) into 
  // a single tab-delimited string to be output to our destination text file.
  .pipe(transform(function(row) {
    return row.join('\t') + '\r';
  }))
  // And finally, output those strings to our destination file.
  .pipe(fs.createWriteStream('./cross_sorted.txt'));

Answer 4

DB 和 sort-stream 是很好的解决方案，但是 DB 可能有点矫枉过正，我认为 sort-stream 最终只是将整个文件排序到内存数组中（在 through结束回调），所以我认为与原始解决方案相比，性能大致相同。
（但我没有运行任何基准，所以我可能错了）。

所以，只是为了破解它，我会提出另一个解决方案:)

---

编辑： 我很想知道这会有多大的不同，所以我运行了一些基准测试。

结果甚至让我感到惊讶，事实证明 sort -k3,3 解决方案比原来的解决方案（简单的数组排序）要好得多，快 10 倍，而 nedb 和sort-stream 解决方案至少比原始解决方案慢 x18 倍（即至少比 sort -k3,3 慢 x180 倍）。

（请参阅下面的基准测试结果）

---

如果在 *nix 机器（Unix、Linux、Mac 等）上，您可以简单地使用
sort -k 3,3 yourInputFile > op_rev.txt 并让操作系统为您进行排序。
您可能会获得更好的性能，因为排序是本机完成的。

或者，如果你想在 Node 中处理排序后的输出：

var util = require('util'),
    spawn = require('child_process').spawn,
    sort = spawn('sort', ['-k3,3', './test.tsv']);

sort.stdout.on('data', function (data) {
    // process data
    data.toString()
        .split('\n')
        .map(line => line.split("\t"))
        .forEach(record => console.info(`Record: ${record}`));
});

sort.on('exit', function (code) {
    if (code) {
        // handle error
    }

    console.log('Done');
});

// optional
sort.stderr.on('data', function (data) {
    // handle error...
    console.log('stderr: ' + data);
});

希望这会有所帮助:)

---

编辑：添加一些基准测试细节。

我很想知道这会有多大的不同，所以我运行了一些基准测试。

以下是结果（在 MacBook Pro 上运行）：

• sort1 使用一种直接的方法，将记录排序在in-memory array 中。
  平均时间：35.​​6s（基线）

• sort2 使用 sort-stream，正如 Joe Krill 所建议的那样。
  平均时间：11.1m（大约 x18.7 倍慢）
  （我不知道为什么。我没有深入研究。）

• sort3 按照 Tamas Hegedus 的建议使用 nedb。
  时间：约16m（约慢27倍）

• sort4 仅通过在终端中执行 sort -k 3,3 input.txt > out4.txt 进行排序
  平均时间：1.2s（大约 x30 倍快）

• sort5 使用sort -k3,3，并处理发送到标准输出的响应
  平均时间：3.65s（大约 x9.7 倍快）