AWS Firehose 到 Elastic Search - 将一条 Firehose 记录转换为多个 Elastic 条目

Question

我的 Lambda 函数中有一个经过优化（标准化）的 JSON 字符串。它被转发到 Firehose 到 Elastic Search。 我的计划是使用 Kinesis Data Transformation Lambda（请参阅https://docs.aws.amazon.com/firehose/latest/dev/data-transformation.html）对 JSON 进行非规范化，以从中获取正确的 Elastic Search 条目。 这个问题是关于可行性/如何做的。

这是我的基本 AWS 设置：

1. Lambda 函数：从 Internet 获取规范化的 JSON，对其进行验证，附加一些属性，然后通过 putRecord 将其转发到 Kinesis Firehose。
2. Kinesis Firehose 转换 lambda：获取一条记录，读取 JSON 并从中生成多个项目，然后将其返回给 Firehose，后者将其转发给 Elastic Search

我的问题是：Firehose/转换 lambda 是否有可能从一条记录中创建多个 Elastic Search 条目？ 我将尝试使用一些伪代码来可视化场景：

lambda.js

exports.handler = async function (event) {
    // 1. get inputDoc from request, which contains multiple es_documents
    // 2. attach timestamp as property
    // result:
    const inputDoc = {
        request_logged_at: '2017-02-07T15:13:01.39256Z',
        es_documents: [
            {
                foo: 'bar'
            },
            {
                foo: 'baz'
            }
        ]
    };

    const firehoseParams = {
        DeliveryStreamName: 'Delivery-Stream',
        Record: {
            Data: JSON.stringify(inputDoc)
        }
    };

    await firehose.putRecord(firehoseParams).promise();
    return { statusCode: 201 };
}

firehose/transform_lambda.js

exports.handler = (event, context) => {
    const record = event.records[0];
    const myDoc = (new Buffer(record.data, 'base64')).toString('utf8');

    // Denormalize request_logged_at into the single documents, so every document in Elastic knows when it got logged
    const docsToElastic = myDoc.es_documents.map(doc => {
        return doc.request_logged_at = myDoc.request_logged_at;
    });

    // This is the main point: How to return this array back to Firehose so Elastic creates multiple table entries?
    // My first guess is the following, as I've seen this syntax in other places 
    // (see first "Note" on this page[https://docs.aws.amazon.com/firehose/latest/dev/record-format-conversion.html])
    const result = docsToElastic.reduce((accumulator, doc) => {
        return accumulator + JSON.stringify(doc);
    }, '');

    // result: '{"foo":"bar","request_logged_at":"2017-02-07T15:13:01.39256Z"}{"foo":"baz","request_logged_at":"2017-02-07T15:13:01.39256Z"}'

    const payload = (new Buffer(result, 'utf8')).toString('base64');
    return {
        recordId: record.recordId,
        result: 'Ok',
        data: payload
    };
}

任何人都可以分享他对这个用例的了解吗？这行得通吗？

顺便说一句：我知道我可以在第一个 lambda 中进行非规范化并使用 firehose.putRecordBatch()，但第一个 lambda 已经有很多任务，这也是分离关注点的问题

Answer 1

我自己解决了。来自docs 的这些提示给了我一个想法：

然后，它 [Firehose] 会生成一个 Elasticsearch 批量请求，以将多条记录索引到您的 Elasticsearch 集群。在将记录发送到 Kinesis Data Firehose 之前，请确保您的记录采用 UTF-8 编码并展平为单行 JSON 对象。

所以实际上在后台发生的是 Firehose 执行了ES bulk request。那么 Firehose 可能会做什么：

    POST _bulk
    { "index" : { "_index" : "my-index" } }
    [FirehoseRecord.Data]
    { "index" : { "_index" : "my-index" } }
    [FirehoseRecord.Data]
    { "index" : { "_index" : "my-index" } }
    [FirehoseRecord.Data]

所以我所做的就是修改 one FirehoseRecord.Data 以拥有多行我的文档，每行由 { "index" : ...} 对象分隔：

const result = docsToElastic.reduce((accumulator, log, currentIndex) => {
    return accumulator +
        // Skip index at first time bcs. that is what Firehose does
        (currentIndex === 0 ? '' : '{ "index" : { "_index" : "my-index"} }' + '\n') +
        JSON.stringify(log) + '\n';
}, '');

输出：

{"foo":"bar","request_logged_at":"2017-02-07T15:13:01.39256Z"}
{ "index" : { "_index" : "my-index" } }
{"foo":"bar","request_logged_at":"2017-02-07T15:13:01.39256Z"}
{ "index" : { "_index" : "my-index" } }
{"foo":"bar","request_logged_at":"2017-02-07T15:13:01.39256Z"}

注意第一个错误的{ "index" : ...}对象。这是因为 Firehose 在整个记录之前添加了第一个 { "index" : ...}。

这已经过测试并且现在可以工作了。