如何在弹性搜索中更新之前检查属性？答案

【问题标题】：How to check properties before update in elasticsearch?如何在弹性搜索中更新之前检查属性？
【发布时间】：2020-07-21 16:22:21
【问题描述】：

我已经看过官方文档了，没办法。

我到 es 的数据来自 kafka，有时可能会出现故障。过去，来自 kafka 的消息被解析并直接插入或更新具有特定 ID 的 ES doc。为了避免旧数据覆盖新数据，我必须检查具有特定 ID 的文档是否已经存在，并且该文档的某些属性是否满足条件。然后我执行 UPDATE 操作（或 INSERT）。

我现在做的是“更新前搜索”。

在更新文档之前，我从 ES 中搜索特定 ID（包含在 kafka msg 中）。然后检查这个文档是否满足条件（例如，update_time是否更旧？）。最后我更新了文档。我将refresh 设置为true 以立即更新索引。

我在担心什么？

似乎是事务性的。

如果只有一个线程同步执行，是否有可能在我处理下一条消息时，在上一条消息过程中更新的文档在 ES 中没有刷新？
如果我有多个线程消耗 kafka 消息，更新前如何检查？我可以使用脚本来解决这个问题吗？

【问题讨论】：

您能解释一下如何让 Kafka 发出“乱序”消息吗？如果与给定文档有关的所有消息都属于同一个分区（它们应该属于），那么您不太可能收到乱序的消息。
消息来自多个生产者。
是的，但是，所有具有相同 ID 的文档都应该放在同一个分区中（即由文档 ID 键入）
但在我的情况下，我使用process rate 之类的东西来识别他们的订单，ID 是一个随机字符串。
那你怎么定义文档的ID呢？

标签： go elasticsearch apache-kafka transactions

【解决方案1】：

如果只有一个线程同步执行，是否有可能在我处理下一条消息时，在上一条消息过程中更新的文档在 ES 中没有刷新？

这是有可能的，因为索引每秒刷新一次（默认情况下），因此既不建议降低此值，也不保证会为您提供所需的结果，因为 Elasticsearch 不是为此而设计的。

如果我有多个线程消耗 kafka 消息，更新前如何检查？我可以使用脚本来解决这个问题吗？

如果要更新的字段数量非常有限，您可以使用脚本。就我个人而言，我发现脚本最适合单字段更新，而且对于角落用例也是如此，它不应该被用作一般做法。除此之外，您将面临与 RDBMS 世界中的存储过程相同的风险。它使数据管理总体上不稳定，并且从长远来看更难以维护/扩展系统。

您的用例最适合 Elasticsearch 开箱即用的乐观锁定支持。查看Elasticsearch Versioning Support了解完整详情。

您可以很好地使用内置文档版本如果并发是您需要解决的唯一问题。但是，如果您需要的不仅仅是并发（乱序消息传递和相应的 ES 更新），那么您应该使用您的应用程序/域特定字段，因为内置版本无法按原样工作。

您可以很好地将任何应用程序特定（数字）字段用作版本字段，并在文档更新期间将其用于乐观锁定。如果您使用这种方法，请特别注意该索引的所有插入、更新、删除操作。从版本支持引用 AS-IS - 使用外部版本控制时，请确保始终将当前版本（和 version_type）添加到任何索引、更新或删除调用。如果您忘记了，Elasticsearch 将使用其内部系统来处理该请求，这将导致版本错误地递增

我建议您先评估内置版本，如果它满足您的需求，请使用它。它将使整体设计更加简单。如果内置版本不符合您的要求，请考虑将应用特定版本作为第二个选项。

【讨论】：

谢谢。我已经采纳了您的建议并使用乐观锁定。它现在完美无缺。 :-)

【解决方案2】：

如果只有一个线程同步执行，是否有可能在我处理下一条消息时，上一条消息处理中更新的文档没有在 ES 刷新？

广告 1. 可以在 ElasticSearch 中保存数据，并在收到陈旧结果后（索引更新之前）的短时间内保存数据

如果我有多个线程消耗 kafka 消息，更新前如何检查？我可以使用脚本来解决这个问题吗？

Ad 2. 如果你在多个线程中处理 Kafka 消息，最好使用业务数据（例如一些业务 id）作为 Kafka 中的分区键，以确保数据按顺序处理。记住要使用Kafka在多线程中消费消息，不要通过单个消费者消费消息稍后扇出到多个线程。

似乎最好确保数据按顺序处理，然后放弃检查 Elasticsearch，因为不能保证给出有效结果。

【讨论】：