Elasticsearch：无序处理更新答案

【问题标题】：Elasticsearch: Handling updates out of orderElasticsearch：无序处理更新
【发布时间】：2019-08-15 18:27:31
【问题描述】：

假设我有以下文件：

{
  "name": "Foo"
  "age": 0
}

我们收到触发这些字段更新的事件：

Event 1
{
  "service_timestamp": "2019-09-15T09:00:01",
  "updated_name": "Bar"
}

Event 2
{
  "service_timestamp": "2019-09-15T09:00:02",
  "updated_name": "Foo"
}

Event 2 比Event 1 晚 1 秒由我们的服务发布，因此我们希望我们的文档首先将“name”属性更新为“Bar”，然后再更新为“Foo”。但是，想象一下，无论出于何种原因，这些事件都会出现故障（Event 2 THEN Event 1）。文档的最终状态将是“Bar”，这不是所需的行为。

我们需要保证我们按照事件上“service_timestamp”字段的顺序更新我们的文档。

我们提出的一个解决方案是在每个字段上添加一个 last_updated_property，如下所示：

{
  "name": {
    "value": "Foo",
    "last_updated_time": 1970-01-01T00:00:00
  }

  "age": {
    "value": 0,
    "last_updated_time": 1970-01-01T00:00:00
  }
}

只有当事件的service_timestamp 发生在文档中属性的last_updated_time 之后时，我们才会更新属性：

{
  "script": {
    "source": "if (ctx._source.name.last_updated_time < event.service_timestamp) { 
                 ctx._source.name.value = event.updated_name;
                 ctx._source.name.last_updated_time = event.service_timestamp;
               }"
  }
}

虽然这可行，但在每次更新时执行读取然后写入似乎代价高昂。还有其他方法可以保证事件以正确的顺序更新吗？

编辑 1：需要考虑的其他一些事项

我们不能假设乱序事件会在很小的时间窗口内发生。想象一下：我们试图更新一个客户的名字，但是这个更新失败了，所以我们将更新事件存储在一些死信队列中，以便稍后重新触发它。我们修复了导致更新失败的错误，并重新触发死信队列中的所有事件。如果在我们修复此错误期间没有发生更新名称字段的更新，则死信队列中的事件应该成功更新属性。但是，如果某些事件确实更新了名称，死信队列中的事件不应更新属性。

【问题讨论】：

您存储在 Elasticsearch 中的每个文档都有一个关联的版本号。该版本号是介于 1 和 2 ^63-1（含）之间的正数。当您第一次为文档建立索引时，它会获取版本 1，并且对该文档的每次写入操作，无论是索引、更新还是删除，Elasticsearch 都会将版本增加 1，因此您应该查看而不是时间戳。看看“Elasticsearch 版本控制系统”。
@TunaMcFish 我怎样才能使用版本号？每次更新都创建一个新版本，但将某种“正在使用”的版本设置为最新的更新时间戳？
当您 GET 一个文档时，您会在响应中返回一个 _version 字段。现在您只需将该版本号添加到您的 POST 中，Elasticsearch 要做的就是比较两个版本号，如果版本相同，那么更新将成功完成，您将得到200 OK 否则（版本不同）elastic 将不会执行该操作，它会用309 CONFLICT 向您发送信号。现在，如果我正确理解了您的问题，我不明白为什么您仍然必须使用自定义的“更新时间戳”字段。
如果您的服务针对给定文档的同一版本触发了两个更新事件，那么第一个会成功，但第二个肯定会失败。
@TunaMcFish 啊，这是一个有趣的想法。我看到的唯一问题是如果我想修改名称，然后快速连续修改年龄字段怎么办？理论上，名称更新会增加版本，导致年龄更新失败。但是，我希望两者都能成功，因为这些事件不会引起任何形式的冲突

标签： elasticsearch

【解决方案1】：

Mousa 所说的一切都是正确的“内部”版本控制，这是您让 Elasticsearch 处理增加版本的地方。

不过，Elasticsearch 还支持“外部”版本控制，您可以在其中为每个更新提供一个版本，以根据当前文档的版本进行检查。我相信这将解决您的事件索引到 ES“无序”的情况，并将在任何事件时间范围内防止这些问题（无论是相隔 1 秒还是 1 周，如您的死信队列示例）。

为此，您需要跟踪主数据存储中的文档版本（Elasticsearch 绝不应该是主数据存储！），并将其附加到索引请求中。

首先你要创建你想要的任何版本号的文档，让我们从 1 开始：

POST localhost:9200/my-index/my-type/<doc id>?version=1&version_type=external -d
{
  "name": "Foo"
  "age": 0
}

然后更新也会从您的服务和/或主数据存储中获得分配的版本

Event 1
POST localhost:9200/my-index/my-type/<doc id>?version=2&version_type=external -d
{
  "service_timestamp": "2019-09-15T09:00:01",
  "updated_name": "Bar"

}

Event 2
POST localhost:9200/my-index/my-type/<doc id>?version=3&version_type=external -d
{
  "service_timestamp": "2019-09-15T09:00:02",
  "updated_name": "Foo"
}

这可确保即使更新应用无序，最新的更新也会获胜。如果在事件 2 之后应用事件 1，您将收到一个代表 VersionConflictEngineException 的 409 错误代码，最重要的是，事件 1 不会覆盖事件 2。

您可以选择将时间戳转换为纪元毫秒并将其作为版本提供，而不是每次将版本 int 递增 - 类似于您创建 last_updated_property 字段的想法，但利用 Elasticsearch 的内置版本控制。这样，最新的时间戳更新将始终“获胜”并最后应用。

我高度建议您阅读这篇关于 Elasticsearch 版本控制的简短博文 - 它比我在这里所做的更详细：https://www.elastic.co/blog/elasticsearch-versioning-support。

祝您搜索愉快！

【讨论】：