Nodejs流CSV，在postgres中创建表并使用COPY FROM导入答案

【问题标题】：Nodejs stream CSV, create table in postgres and import using COPY FROMNodejs流CSV，在postgres中创建表并使用COPY FROM导入
【发布时间】：2014-04-03 13:49:18
【问题描述】：

有没有一种好方法可以将流形式的大型 CSV (4GB+) 放入 NodeJS 中的 postgres 数据库中？

特别是，我想取第一行（标题行）并从中制作一个创建查询。然后，我想将文件的其余部分流式传输到 from 语句的副本。

如果我只想发送到复制命令，这将是可行的，例如：

function copyStreamIntoTable (inputStream) {
  var deferred = Q.defer();
  pg.connect("pg://postgres@localhost/npi_demo", function (err, client) {
    var s = client.copyFrom("COPY hptc (code, type, classification, specialization, definition, notes) FROM STDIN WITH CSV HEADER");

    inputStream.pipe(through(function (data) {
      this.queue(data.toString("ascii"));
    })).pipe(s).on('close', function () {
      deferred.resolve();
    });
  });

  return deferred.promise;
}

但我希望流读取第一行，然后运行创建查询。创建查询完成后，id 喜欢将流的其余部分流式传输到 copyFrom。

有没有一种优雅的方法来做到这一点？对于 nodejs 和流，我仍然是一个相对较新的人。

【问题讨论】：

检查 node.js 的 PostgreSQL 接口是否在其 API 中明确支持COPY。如果可能，您将希望使用任何此类支持，而不是尝试直接使用 COPY 语句。
@CraigRinger 谢谢——我会检查一下。
作为旁注，我对节点中的流感到非常困惑......我知道我可以使用 readline 之类的东西来阅读第一行......我可能会把剩下的写到使用 readline 的 postgres 流......但是为了正确尊重 pg 施加的背压，除非我使用管道，否则事情似乎很快变得更加复杂
@MichaelWasser 嘿，我是 node 新手，我有同样的问题，你能把代码发给我吗？或者更多解释一下这个

标签： node.js postgresql csv

【解决方案1】：

对我的问题想出了一个更笼统的答案：

var fs = require('fs'),
    through = require('through'),
    split = require('split');

var inFile = fs.createReadStream('./lines'),
    outFile = fs.createWriteStream('./out'),
    headers;

var th = through(function (data) {
  if (typeof headers === "undefined") {
    headers = data;
    th.pause();
    setTimeout(function () { th.resume(); }, 5000);
  } else {
    this.queue(data + "\n");
  }
});

inFile.pipe(split())
  .pipe(th)
  .pipe(outFile)
  .on("close", function () {
    console.log("had headers: " + headers);
  });

npm 安装的流助手在哪里 through 和 split。

如果inFile的内容是：

one
two
three

输出将是 had headers: one 并且 outFile 将包含内容

two
three

在我的具体问题的上下文中 - 超时将被查询数据库以创建表和写入输出文件将被替换为写入 copyFrom 流。

【讨论】：

酷，谢谢。您知道对于 CSV 的大小是否有任何建议？你是把你的文件分成几个 csv 字符串还是一个？有recos吗？
没有具体的拆分方式——我使用这种方法导入了一个 4.5GB 的 csv