AWS Lambda 和 Node：在流式传输时写入数据 - 过早结束并且数据丢失答案

【问题标题】：AWS Lambda & Node: Write data while streaming - ends prematurely and data is missingAWS Lambda 和 Node：在流式传输时写入数据 - 过早结束并且数据丢失
【发布时间】：2019-12-14 01:56:52
【问题描述】：

我有一个由写入 S3 存储桶触发的 Lambda 函数。它读取写入存储桶的 JSON 文件，解析出各个记录，然后将它们写入数据库。

问题是；我不确定自己做错了什么，因为流结束并且 Lambda 在写入所有数据之前就退出了。

我的可读流处于“流动模式”，在数据库写入期间我正在暂停/恢复。根据文档，这应该可以解决问题，但它没有按预期工作。

Lambda 处理程序：

exports.handler = async (event, context) => {
    let result = false;
    try {
        result = await parseData(event);
    } catch (e) {
        console.error(e);
    }
    return result;
};

承诺：

const StreamArray = require("stream-json/streamers/StreamArray");

async parseData(event) {
    try {
        let objectStream = s3.getObject(params).createReadStream();
        const streamParser = StreamArray.withParser();
        return new Promise((resolve, reject) => {
            objectStream.pipe(streamParser).on("data", async streamData => {
                objectStream.pause();
                let result = await writeData(streamData);
                objectStream.resume();
            }).on("finish", () => {
                console.log("STREAM FINISH!");
                resolve(true);
            }).on("error", e => {
                console.error("Stream error:", e);
                reject(e);
            });
        });
    } catch (e) {
        console.error(e);
    }
}

【问题讨论】：

JSON 文件有多大？你真的需要流式传输吗？
@jarmod 它可能非常大，我无法预测大小，所以应该流式传输。

标签： javascript node.js amazon-web-services amazon-s3 node-streams

【解决方案1】：

通过简单地将 stream-json 换成 JSONStream 来实现它，无论如何这是一个更广泛使用的包。现在像魅力一样工作！

const JSONStream = require("JSONStream");

async parseData(event) {
    try {
        let objectStream = s3.getObject(params).createReadStream();
        const streamParser = JSONStream.parse("*");
        return new Promise((resolve, reject) => {
            objectStream.pipe(streamParser).on("data", async streamData => {
                streamParser.pause();
                let result = await writeData(streamData);
                streamParser.resume();
            }).on("finish", () => {
                console.log("STREAM FINISH!");
                resolve(true);
            }).on("error", e => {
                console.error("Stream error:", e);
                reject(e);
            });
        });
    } catch (e) {
        console.error(e);
    }
}

【讨论】：