Kinesis Stream 和 Kinesis Firehose 更新 Elasticsearch 索引

Posted 2023-02-24

【中文标题】Kinesis Stream 和 Kinesis Firehose 更新 Elasticsearch 索引

【英文标题】：Kinesis Stream and Kinesis Firehose Updating Elasticsearch Indexes

【发布时间】：2019-01-01 06:48:37

【问题描述】：

我们想使用 kinesis 流和 firehose 来更新 aws 托管的 elasticsearch 集群。我们有数百个不同的索引（对应于我们的数据库分片）需要更新。创建 firehose 时，它​​要求我指定要更新的特定索引名称。这是否意味着我需要为集群中的每个索引创建一个单独的 firehose？或者有没有办法配置 firehose，以便它知道根据数据内容使用什么索引。

此外，我们将有 20 个左右的独立生产者将数据发送到 kinesis 流（这些生产者中的每一个都会为 10 个不同的索引生成数据）。我是否还需要为每个生产者提供单独的运动流。

总结： 20 个生产者（EC2 实例）-> 每个生产者将 20 个不同索引的数据发送到 kinesis 流 -> 然后，kinesis 流使用 firehose 更新其中包含 200 个索引的单个集群。

注意：所有索引都具有相同的映射和名称 Temple，即 index_1、index_2...index_200

编辑：当我们重新索引数据时，我们会沿着 index_1-v2 的行创建新索引。显然，我们不想在创建每个索引版本时为它们创建一个新的 firehose。新的索引名称可以包含在发送到 kinesis 流的 JSON 中。

【参考方案1】：

如您所料，Firehose 是解决此问题的错误解决方案，至少如前所述。它专为流（不是生产者！）和索引之间存在 1:1 对应关系的情况而设计。诸如点击流数据或日志聚合之类的东西。

对于任何解决方案，您都需要提供一种机制来识别记录所属的索引。您可以通过为每种消息类型创建一个单独的 Kinesis 流来实现这一点（在这种情况下，您可以使用 Firehose），但这意味着您的生产者必须决定将每条消息写入哪个流。这可能会给您的生产者带来不必要的复杂性，并且还可能会增加您无法接受的成本。

因此，假设您想要所有消息的单个流，您需要一个消费者应用程序和某种方法来对这些消息进行分组。您可以在记录本身中包含消息类型（/索引名称），或为此目的使用分区键。分区键使实现更容易一些，因为它保证相同索引的记录将存储在同一个分片上，但这意味着您的生产者可能会受到限制。

对于消费者，您可以使用在 EC2 上运行的始终在线应用程序，或者使用stream invoke a Lambda function。

如果您使用分区键来识别消息类型，则使用 Lambda 非常好，因为每次调用只查看单个分片（调用中可能仍有多个分区键）。不利的一面是，Lambda 将每秒轮询一次流，如果您有多个流消费者，这可能会导致限制（使用独立应用程序，您可以控制它轮询流的频率）。

【讨论】：

这不是答案。

能否让生产者（EC2 实例）检查指定索引名称的消防软管是否存在，如果不存在，则针对该索引创建一个新的消防软管？数据是否可以直接从 EC2 实例直接传输到 Firehose，或者它们之间是否需要有 kinesis 数据流？

@Rohit.007 - 那你有更好的吗？或者至少有一个理由不这样做？

@CorribView - 你可以这样做，但它为制作人提供了大量的知识。 Firehose 接受的吞吐量也受到限制（与 Kinesis Stream 不同，它不易扩展）。

谢谢。我看到每个帐户每个区域只能有 50 个数据流（假设包括 firehose 流）所以我想这不会像我们需要的那样工作，如果创建新版本索引，可能会有 400 个流：docs.aws.amazon.com/firehose/latest/dev/limits.html