创建/更新文档日期的摄取管道

Question

我正在尝试实现类似于 Mysql 的行为，为我通过 ES 管道索引的每个文档的元数据添加 insert_at/updated_at 。

我的管道是这样的：

{
  "description": "Adds createdAt and updatedAt style timestamps",
  "processors": [
    {
      "set": {
        "field": "_source.indexed_at",
        "value": "{{_ingest.timestamp}}",
        "override": false
      }
    },
    {
      "set": {
        "field": "_source.updated_at",
        "value": "{{_ingest.timestamp}}",
        "override": true
      }
    }
  ]
}

我没有映射，只尝试添加一个文档：

POST test_pipelines/doc/1?pipeline=timestamps
{
  "foo": "bar"
}

管道成功创建indexed_at并且updated_at：

{
  "_index": "test_pipelines",
  "_type": "doc",
  "_id": "1",
  "_score": 1,
  "_source": {
    "indexed_at": "2018-07-12T10:47:27.957Z",
    "updated_at": "2018-07-12T10:47:27.957Z",
    "foo": "bar"
  }
}

但是，如果我尝试更新 doc 1，indexed_at该字段每次都会更改为更新文档的同一日期。

更新请求示例：

POST test_pipelines/doc/1?pipeline=timestamps
{
  "foo": "bor"
}

有什么方法可以告诉处理器不要更新indexed_at字段吗？

Answer 1

发生这种情况的原因是set处理器只会在您发送的文档的上下文中运行，而不是在存储的文档（如果有）的上下文中运行。因此，override在这里不起作用，因为您发送的文档既不包含indexed_at也不包含updated_at，这就是每次调用时都设置这两个字段的原因。

当您PUT第二次编辑文档时，您并没有更新它，而是实际上从头开始重新索引它（即覆盖您发送的第一个版本）。摄取管道不适用于更新操作。例如，如果您尝试以下更新调用，它将失败。

POST test_pipelines/doc/1/_update?pipeline=timestamps
{
  "doc": {
    "foo": "bor"
  }
}

如果您想坚持使用摄取管道，使其发挥作用的唯一方法是GET先查看文档，然后更新所需的字段。例如，

# 1. index the document the first time
PUT test_pipelines/doc/1?pipeline=timestamps
{
  "foo": "bar"
}

# 2. GET the indexed document
GET test_pipelines/doc/1

# 3. update the foo field and index it again
PUT test_pipelines/doc/1?pipeline=timestamps
{
  "indexed_at": "2018-07-20T05:08:52.293Z",
  "updated_at": "2018-07-20T05:08:52.293Z",
  "foo": "bor"
}

# 4. When you GET the document the second time, you'll see your pipeline worked
GET test_pipelines/doc/1

这将返回：

{
  "indexed_at": "2018-07-20T05:08:52.293Z",
  "updated_at": "2018-07-20T05:08:53.345Z",
  "foo": "bor"
}

我绝对同意这确实很麻烦，但是我上面给出的链接列举了更新操作不支持管道的所有原因。

另一种让它按照您喜欢的方式工作（没有管道）的方法是使用脚本化的 upsert 操作（其工作方式类似于上面的步骤 2 和 3，即在单个原子操作中 GET 和 PUT 文档），这也可以工作与您的批量通话。基本上是这样的。首先，您需要存储一个脚本，您将调用该脚本来执行索引和更新操作：

POST _scripts/update-doc
{
  "script": {
    "lang": "painless",
    "source": "ctx._source.foo = params.foo; ctx._source.updated_at = new Date(); if (ctx._source.indexed_at == null) ctx._source.indexed_at = ctx._source.updated_at;"
  }
}

然后，您可以像这样第一次索引您的文档：

POST test_pipelines/doc/1/_update
{
  "script": {
    "id": "update-doc",
    "params": {
      "foo": "bar"
    }
  },
  "scripted_upsert": true,
  "upsert": {}
}

索引文档将如下所示：

{
    "updated_at": "2018-07-20T05:57:40.510Z",
    "indexed_at": "2018-07-20T05:57:40.510Z",
    "foo": "bar"
}

您可以在更新文档时使用完全相同的调用：

POST test_pipelines/doc/1/_update
{
  "script": {
    "id": "update-doc",
    "params": {
      "foo": "bor"             <--- only this changes
    }
  },
  "scripted_upsert": true,
  "upsert": {}
}

更新后的文档将如下所示，正是您想要的：

{
    "updated_at": "2018-07-20T05:58:42.825Z",
    "indexed_at": "2018-07-20T05:57:40.510Z",
    "foo": "bor"
}