AWS Kinesis Firehose lambda 如何向 ElasticSearch 发送更新和删除请求？

Question

我没有看到 AWS Kinesis Firehose lambda 如何向 ElasticSearch（AWS OpenSearch 服务）发送更新和删除请求。

Elasticsearch 文档 API 提供 CRUD 操作： https://www.elastic.co/guide/en/elasticsearch/reference/current/docs.html

我找到的示例涉及 Create 案例，但没有说明如何执行 delete 或 update 请求。 https://aws.amazon.com/blogs/big-data/ingest-streaming-data-into-amazon-elasticsearch-service-within-the-privacy-of-your-vpc-with-amazon-kinesis-data-firehose/ https://github.com/amazon-archives/serverless-app-examples/blob/master/python/kinesis-firehose-process-record-python/lambda_function.py

示例中的输出格式未显示指定 create、update 或 delete 请求的方法：

       output_record = {
           'recordId': record['recordId'],
           'result': 'Ok',
           'data': base64.b64encode(payload)
       }

除了这些示例，我没有找到 kinesis firehose lambda 处理程序应该的输出格式的定义 return。

Answer 1

在您的案例 OpenSearch(ES) 中，Firehose 在将记录传送到目的地之前使用 lambda 函数对其进行转换，因此它们仅用于修改数据的结构，但不能用于影响 CRUD 操作。 Firehose 只能将记录插入特定索引。如果您需要一个简单的选项来在一段时间后从 ES 索引中删除记录，请在为您的 Firehose 流指定目标时查看“索引轮换”选项。

如果您想对 ES 使用 CRUD 操作并继续使用 Firehose，我建议以原始格式将记录发送到 S3 存储桶，然后在对象上传事件上触发 lambda 函数，该函数将根据字段执行 CRUD 操作在你的有效负载中。

从 lambda 对 ES 执行 CRUD 操作的一个很好的例子 https://github.com/chankh/ddb-elasticsearch/blob/master/src/lambda_function.py

这个特定示例旨在将数据从 DynamoDB 流发送到 ES，但它对您来说应该是一个很好的起点

Answer 2

Firehose 只能插入到 ES 中。它不能进行更新和删除。如果您想执行更新和删除，您将需要编写一个从 Kinesis 流上的事件中触发的 lambda

https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html

AWS Kinesis Firehose lambda 如何向 ElasticSearch 发送更新和删除请求？

How can AWS Kinesis Firehose lambda send update and delete requests to ElasticSearch?

amazon-web-services

elasticsearch

aws-lambda

amazon-kinesis-firehose