更新时的批量操作

就像mget允许我们一次性检索多个文档一样，bulk API允许我们使用单一请求来实现多个文档的create、index、update或delete。这对索引类似于日志活动这样的数据流非常有用，它们可以以成百上千的数据为一个批次按序进行索引。

bulk请求体如下，它有一点不同寻常：

  1. { action: { metadata }}\n
  2. { request body        }\n
  3. { action: { metadata }}\n
  4. { request body        }\n
  5. ...

这种格式类似于用"\n"符号连接起来的一行一行的JSON文档流(stream)。两个重要的点需要注意：

  • 每行必须以"\n"符号结尾，包括最后一行。这些都是作为每行有效的分离而做的标记。

  • 每一行的数据不能包含未被转义的换行符，它们会干扰分析——这意味着JSON不能被美化打印。

提示:

在《批量格式》一章我们介绍了为什么bulk API使用这种格式。

action/metadata这一行定义了文档行为(what action)发生在哪个文档(which document)之上。

行为(action)必须是以下几种：

行为 解释
create 当文档不存在时创建之。详见《创建文档》
index 创建新文档或替换已有文档。见《索引文档》和《更新文档》
update 局部更新文档。见《局部更新》
delete 删除一个文档。见《删除文档》

在索引、创建、更新或删除时必须指定文档的_index、_type、_id这些元数据(metadata)。

例如删除请求看起来像这样：

  1. { "delete": { "_index": "website", "_type": "blog", "_id": "123" }}

请求体(request body)由文档的_source组成——文档所包含的一些字段以及其值。它被index和create操作所必须，这是有道理的：你必须提供文档用来索引。

这些还被update操作所必需，而且请求体的组成应该与update API（doc, upsert, script等等）一致。删除操作不需要请求体(request body)。

  1. { "create":  { "_index": "website", "_type": "blog", "_id": "123" }}
  2. { "title":    "My first blog post" }

如果定义_id，ID将会被自动创建：

  1. { "index": { "_index": "website", "_type": "blog" }}
  2. { "title":    "My second blog post" }

为了将这些放在一起，bulk请求表单是这样的：

  1. POST /_bulk
  2. { "delete": { "_index": "website", "_type": "blog", "_id": "123" }} <1>
  3. { "create": { "_index": "website", "_type": "blog", "_id": "123" }}
  4. { "title":    "My first blog post" }
  5. { "index":  { "_index": "website", "_type": "blog" }}
  6. { "title":    "My second blog post" }
  7. { "update": { "_index": "website", "_type": "blog", "_id": "123", "_retry_on_conflict" : 3} }
  8. { "doc" : {"title" : "My updated blog post"} } <2>
  • <1> 注意delete行为(action)没有请求体，它紧接着另一个行为(action)
  • <2> 记得最后一个换行符

Elasticsearch响应包含一个items数组，它罗列了每一个请求的结果，结果的顺序与我们请求的顺序相同：

  1. {
  2.    "took": 4,
  3.    "errors": false, <1>
  4.    "items": [
  5.       {  "delete": {
  6.             "_index":   "website",
  7.             "_type":    "blog",
  8.             "_id":      "123",
  9.             "_version": 2,
  10.             "status":   200,
  11.             "found":    true
  12.       }},
  13.       {  "create": {
  14.             "_index":   "website",
  15.             "_type":    "blog",
  16.             "_id":      "123",
  17.             "_version": 3,
  18.             "status":   201
  19.       }},
  20.       {  "create": {
  21.             "_index":   "website",
  22.             "_type":    "blog",
  23.             "_id":      "EiwfApScQiiy7TIKFxRCTw",
  24.             "_version": 1,
  25.             "status":   201
  26.       }},
  27.       {  "update": {
  28.             "_index":   "website",
  29.             "_type":    "blog",
  30.             "_id":      "123",
  31.             "_version": 4,
  32.             "status":   200
  33.       }}
  34.    ]
  35. }}
  • <1> 所有子请求都成功完成。

每个子请求都被独立的执行，所以一个子请求的错误并不影响其它请求。如果任何一个请求失败，顶层的error标记将被设置为true，然后错误的细节将在相应的请求中被报告：

  1. POST /_bulk
  2. { "create": { "_index": "website", "_type": "blog", "_id": "123" }}
  3. { "title":    "Cannot create - it already exists" }
  4. { "index":  { "_index": "website", "_type": "blog", "_id": "123" }}
  5. { "title":    "But we can update it" }

响应中我们将看到create文档123失败了，因为文档已经存在，但是后来的在123上执行的index请求成功了：

  1. {
  2.    "took": 3,
  3.    "errors": true, <1>
  4.    "items": [
  5.       {  "create": {
  6.             "_index":   "website",
  7.             "_type":    "blog",
  8.             "_id":      "123",
  9.             "status":   409, <2>
  10.             "error":    "DocumentAlreadyExistsException <3>
  11.                         [[website][4] [blog][123]:
  12.                         document already exists]"
  13.       }},
  14.       {  "index": {
  15.             "_index":   "website",
  16.             "_type":    "blog",
  17.             "_id":      "123",
  18.             "_version": 5,
  19.             "status":   200 <4>
  20.       }}
  21.    ]
  22. }
  • <1> 一个或多个请求失败。
  • <2> 这个请求的HTTP状态码被报告为409 CONFLICT。
  • <3> 错误消息说明了什么请求错误。
  • <4> 第二个请求成功了，状态码是200 OK。

这些说明bulk请求不是原子操作——它们不能实现事务。每个请求操作时分开的，所以每个请求的成功与否不干扰其它操作。

不要重复

你可能在同一个index下的同一个type里批量索引日志数据。为每个文档指定相同的元数据是多余的。就像mget API，bulk请求也可以在URL中使用/_index或/_index/_type:

  1. POST /website/_bulk
  2. { "index": { "_type": "log" }}
  3. { "event": "User logged in" }

你依旧可以覆盖元数据行的_index和_type，在没有覆盖时它会使用URL中的值作为默认值：

  1. POST /website/log/_bulk
  2. { "index": {}}
  3. { "event": "User logged in" }
  4. { "index": { "_type": "blog" }}
  5. { "title": "Overriding the default type" }

多大才算太大？

整个批量请求需要被加载到接受我们请求节点的内存里，所以请求越大，给其它请求可用的内存就越小。有一个最佳的bulk请求大小。超过这个大小，性能不再提升而且可能降低。

最佳大小，当然并不是一个固定的数字。它完全取决于你的硬件、你文档的大小和复杂度以及索引和搜索的负载。幸运的是，这个最佳点(sweetspot)还是容易找到的：

试着批量索引标准的文档，随着大小的增长，当性能开始降低，说明你每个批次的大小太大了。开始的数量可以在1000~5000个文档之间，如果你的文档非常大，可以使用较小的批次。

通常着眼于你请求批次的物理大小是非常有用的。一千个1kB的文档和一千个1MB的文档大不相同。一个好的批次最好保持在5-15MB大小间。