使用 Express 接收 JSON 数据块并插入 PostgreSQL

Question

我想在提问之前做一个简短的介绍。

下面的示例非常适合将 400MB JSON 文件作为流读取并将数据附加到 PostgreSQL 数据库中。

var fs = require('fs')
var JSONStream = require('JSONStream')
var es = require('event-stream')
var Pool = require('pg').Pool

var pool = new Pool({user: user, host: host, database: database, password: password, port: port})
var types = require('pg').types
types.setTypeParser(1114, str => str)

var getStream = function () {
    var jsond = 'export.json', //400MB File
        stream = fs.createReadStream(jsond, {encoding: 'utf8'}),
        parser = JSONStream.parse('*');
        return stream.pipe(parser);
};

var i = 0
getStream().pipe(es.mapSync(function (d){
    pool.query(`INSERT INTO metrics ("id","test") values ($1,$2)`,[d.id,d.test]).catch(console.error).then(function(){
        console.log('Inserted...', i++)
    })
}));

我使用下面的代码通过 express 和 express-fileupload 从上传表单中获取 JSON 文件。它也可以完美运行（我省略了通常的标题）：

server.js

app.post('/upload', function(req, res, next){
    if (!req.files || Object.keys(req.files).length === 0) {
        return res.status(400).send('No file');
    }

    let sampleFile = req.files.archive;

    sampleFile.mv(`${pathtoupload}}/upload.js`, function(err) {
        if (err) return res.status(500).send(err);
        res.send('File uploaded!');
    })
})

index.html

<div class="card">
    <div class="card-body">
        <div class="row">
            <div class="col-12 col-sm-12">
                <form method="POST" encType="multipart/form-data">
                    <input type="file" name="archive" />
                    <button type="submit" class="btn btn-success btn-lg"><i class="material-icons">cloud_upload</i> Send</button>
                </form>  
            </div>
        </div>
    </div>
</div>

<script>
        $("form").submit(function(e){
            e.preventDefault()

            var formData = new FormData(this)

            $.ajax({
                url: '/upload',
                type: 'POST',
                data: formData,
                //dataType: 'application/octet-stream',
                cache: false,
                contentType: false,
                processData: false,
                xhr: function() {
                    var myXhr = $.ajaxSettings.xhr();
                    if (myXhr.upload) {
                        myXhr.upload.addEventListener('progress', function(a) {
                            console.log('P',a)
                        }, false);
                    }
                    return myXhr;
                },
                success: function(data) {
                    alert('ok',data)
                },
            });
        });
</script>

问题：如何接收大量 JSON 数据以表示、解析然后附加到数据库？

是否可以，或者我必须等待上传完成？

我尝试了以下但没有成功，服务器控制台中没有打印只是“上传”，但在客户端我可以在控制台中看到正在上传的块

var JSONStream = require('JSONStream')
var es = require('event-stream')
var stream = require('stream');

app.post('/upload', function(req, res){
    console.log('upload')
        
    var readable = new stream.Readable({encoding: 'utf8'})

    req.on('data', function onRequestData(chunk){
        readable.push(chunk)
    })
    req.on('end', function(){
        res.send({})
    })

    var getStream = function () {
        parser = JSONStream.parse('*')
        return readable.pipe(parser)
    }

    getStream().pipe(es.mapSync(function(d){
        console.log(d)
        ///the pg insert part
    }))
})

Answer 1

我建议您为此使用pg-copy-streams。它允许您将二进制数据直接流入和流出表。这正是您所需要的。下面是一个取自 NPM 页面的示例：

从文件到表的管道

    var fs = require('fs');
    var { Pool } = require('pg');
    var copyFrom = require('pg-copy-streams').from;
     
    var pool = new Pool();
     
    pool.connect(function(err, client, done) {
      var stream = client.query(copyFrom('COPY my_table FROM STDIN'));
      var fileStream = fs.createReadStream('some_file.tsv')
      fileStream.on('error', done);
      stream.on('error', done);
      stream.on('end', done);
      fileStream.pipe(stream);
    });