经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 程序设计 » Elasticsearch » 查看文章
elasticsearch6.7 05. Document APIs（7）Update By Query API
来源：cnblogs　　作者：说_不_得　　时间：2019/4/15 9:10:44　　对本文有异议

6、Update By Query API

_update_by_query 接口可以在不改变 source 的情况下对 index 中的每个文档进行更新。这对于获取新属性或其他联机映射更改很有用。以下是 API：

  1. POST twitter/_update_by_query?conflicts=proceed

这将返回如下内容：

  1. {
  2.   "took" : 147,
  3.   "timed_out": false,
  4.   "updated": 120,
  5.   "deleted": 0,
  6.   "batches": 1,
  7.   "version_conflicts": 0,
  8.   "noops": 0,
  9.   "retries": {
  10.     "bulk": 0,
  11.     "search": 0
  12.   },
  13.   "throttled_millis": 0,
  14.   "requests_per_second": -1.0,
  15.   "throttled_until_millis": 0,
  16.   "total": 120,
  17.   "failures" : [ ]
  18. }

_update_by_query 在开始执行的时候获得一个快照，并使用内部版本控制对找到的内容进行索引。这意味着，如果文档在获取快照和处理索引请求之间发生更改，则会出现版本冲突。当版本匹配时，文档会更新，对应的版本号也会递增。

由于内部版本控制不支持将0作为有效版本号，因此无法使用_update_by_query更新版本号为零的文档。

所有的更新和查询失败都将导致_update_by_query 中止，并且在 响应中返回 failures。已经更新的内容仍然存在。也就是说这个操作不会回滚，只会中止。当第一个错误引起中止后，所有的错误都会返回到 failures 元素中，因此有可能有相当多的失败实例。

如何您想在遇到版本冲突时继续执行 _updata_by_query，那么可以在 url 中设置 conflicts=proceed 或是在请求中设置 conflicts":"proceed。第一个例子是这样设置的，因为它想要获得发生改变的在线映射，版本冲突仅仅意味着冲突文档在update_by_query 的开始时间和试图更新文档的时间之间被更新。这没关系，因为该更新将获取联机映射更新。

下面的例子将会 update twitter 索引中的 tweets：

  1. POST twitter/_doc/_update_by_query?conflicts=proceed

您还可以使用 Query DSL 限制 _update_by_query，下面的例子将会更新 twitter 索引中所有 user 字段为 kimchy 的文档：

  1. POST twitter/_update_by_query?conflicts=proceed
  2. {
  3.   "query": { 
  4.     "term": {
  5.       "user": "kimchy"
  6.     }
  7.   }
  8. }

目前，我们在更新文档的时候没有更改 source。这对诸如获取新属性很有用，但只是其中一般的乐趣。_update_by_query 支持用脚本来更新文档。下面的例子将会把所有user为 kimchy 的 likes 字段加 1：

  1. POST twitter/_update_by_query
  2. {
  3.   "script": {
  4.     "source": "ctx._source.likes++",
  5.     "lang": "painless"
  6.   },
  7.   "query": {
  8.     "term": {
  9.       "user": "kimchy"
  10.     }
  11.   }
  12. }

正如在 update API 中一样，您可以设置 ctx.op 来更改执行的操作：

  • noop：如果脚本确定不需要进行任何更改，请设置 ctx.op=noop。这将导致_update_by_query 从其更新中忽略该文档。此无操作将在响应体的 noop 计数器中报告。
  • delete：如果你决定删除该文档，请设置 ctx.op=delete。删除操作会在响应体的 deleted 计数器中报告。

将 ctx.op 设置为其他值都是错误的，在 ctx 中设置其他字段也都是错误的。

请注意，如果我们取消设置 conflicts=proceed。在这种情况下，我们希望版本冲突能够中止该操作以便于我们处理失败的原因。

此API不允许您移动涉及到的文档，只能修改它的 source。这是有意为之的。因为我们不想将文件移动位置。

此API也可以一次修改多个索引和多个类型，如：

  1. POST twitter,blog/_doc,post/_update_by_query

如何设置了routing ，那么操作的分片还应该满足路由的条件，如：

  1. POST twitter/_update_by_query?routing=1

_update_by_query 默认使用的scroll 批次大小是 1000，您可以通过设置 scroll_size来 修改此值：

  1. POST twitter/_update_by_query?scroll_size=100

_update_by_query 也可以指定 Ingest Node接口的 pipeline，如：

  1. PUT _ingest/pipeline/set-foo
  2. {
  3.   "description" : "sets foo",
  4.   "processors" : [ {
  5.       "set" : {
  6.         "field": "foo",
  7.         "value": "bar"
  8.       }
  9.   } ]
  10. }
  11. POST twitter/_update_by_query?pipeline=set-foo

6.1 URL 参数（URL Parameters）

_update_by_query 接口支持的参数有 pretty、 refresh、 wait_for_completion、wait_for_active_shards、timeout、scroll

  • refresh
    • 执行完更新操作后刷新update_by_query 中查询涉及的分片。这和 update API 的 refresh 不同，update API 的 refresh 仅刷新接收到更新请求的分片。_update_by_query 的 refresh 参数不支持 wait_for。
  • wait_for_completion
    • 如果请求中 wait_for_completion=false，那么 elasticsearch 将会执行预检查、执行请求，然后返回可与任务 API 一起使用的任务，以取消或获取任务的状态。elasticsearch 还将在.task/task/${taskId} 处创建此任务的记录文档。您可以根据需要保留或删除该文档。用完删除后，elasticsearch 可以回收其占用的空间。
  • wait_for_active_shards
    • 控制在执行请求前需要有几个可用的分片数。
  • timeout
    • 控制每个请求等待分片从不可用状态到可用状态的时间。因为_update_by_query 使用滚动搜索，所以可以指定 scroll 参数来控制它使 “搜索上下文” 保持活动的时间，例如：scroll=10m。默认情况下，它是 5 分钟。
  • request_per_second
    • 可以设置为任意正数（如 1.4、6、1000 等）。通过设置等待的时间来控制_update_by_query 批量执行更新操作的速率。当 requests_per_second 设置为 - 1 时禁用该控制。

速率限制是通过在批量处理之间等待来完成的，这样就可以为_update_by_query内部使用的回滚指定一个考虑填充的超时时间。填充时间是批处理大小除以 requests_per_second 与写入时间只差。默认情况下，批处理大小为 1000，因此如 request_per_second 设置为 500：

  1. target_time = 1000 / 500 per second = 2 seconds
  2. wait_time = target_time - write_time = 2 seconds - .5 seconds = 1.5 seconds

由于该批处理是作为单个 _bulk 请求发出的，因此大批量的请求将导致ElasticSearch创建多个请求，然后在启动下一个集合之前等待一段时间。这是“突发”而不是“平稳”。默认值为-1。

6.2 响应体（Response body）

  1. {
  2.   "took" : 147,
  3.   "timed_out": false,
  4.   "total": 5,
  5.   "updated": 5,
  6.   "deleted": 0,
  7.   "batches": 1,
  8.   "version_conflicts": 0,
  9.   "noops": 0,
  10.   "retries": {
  11.     "bulk": 0,
  12.     "search": 0
  13.   },
  14.   "throttled_millis": 0,
  15.   "requests_per_second": -1.0,
  16.   "throttled_until_millis": 0,
  17.   "failures" : [ ]
  18. }
  • took
    • 整个操作耗费的毫秒数
  • timed_out
    • 如果在执行 _update_by_query 操作时出现超时，那么这个标识将会返回 true
  • total
    • 成功执行操作的文档的数量
  • updated
    • 成功的更新了多少个文档
  • deleted
    • 成功的删除了多少个文档
  • batches
    • 回滚数
  • verison_conflicts
    • 操作过程中出现版本冲突的数量
  • noops
    • 由于 ctx.op=noop 设置造成的忽略的文档数
  • retries
    • 重复尝试的次数，bulk 是批量更新操作重复尝试的次数，search 是查询的重复尝试次数
  • throthled_millis
    • requests_per_second 参数引起的请求等待时间
  • requests_per_second
    • 在操作过程中，每秒执行的请求数
  • throttled_until_millis
    • 执行_update_by_query时这个值始终0，只在在调用Task API时该值才有意义，它表示下一次（自纪元以来）为了符合requests_per_second将再次执行请求的毫秒数。
  • failures
    • 执行失败的数组，包含在执行过程中任何不可恢复的错误。如果这个数组不是空的，那么请求会因为这些失败而中止。_delete_by_query是使用批处理实现的，任何失败都会导致整个执行被中止。可以使用conflicts参数来防止reindex在版本冲突时造成操作中止。

6.3 结合 taskAPi 使用（Works with the Task API）

您可以使用 Task API 获取任何正在进行 update_by_query 请求的状态：

  1. GET _tasks?detailed=true&actions=*byquery

返回值：

  1. {
  2.   "nodes" : {
  3.     "r1A2WoRbTwKZ516z6NEs5A" : {
  4.       "name" : "r1A2WoR",
  5.       "transport_address" : "127.0.0.1:9300",
  6.       "host" : "127.0.0.1",
  7.       "ip" : "127.0.0.1:9300",
  8.       "attributes" : {
  9.         "testattr" : "test",
  10.         "portsfile" : "true"
  11.       },
  12.       "tasks" : {
  13.         "r1A2WoRbTwKZ516z6NEs5A:36619" : {
  14.           "node" : "r1A2WoRbTwKZ516z6NEs5A",
  15.           "id" : 36619,
  16.           "type" : "transport",
  17.           "action" : "indices:data/write/update/byquery",
  18.           "status" : {    
  19.             "total" : 6154,
  20.             "updated" : 3500,
  21.             "created" : 0,
  22.             "deleted" : 0,
  23.             "batches" : 4,
  24.             "version_conflicts" : 0,
  25.             "noops" : 0,
  26.             "retries": {
  27.               "bulk": 0,
  28.               "search": 0
  29.             },
  30.             "throttled_millis": 0
  31.           },
  32.           "description" : ""
  33.         }
  34.       }
  35.     }
  36.   }
  37. }

status：这个对象包含了当前任务的实际状态。total 字段是本次操作需要重新索引的文档数。你可以通过 updated, created, and deleted 字段估计处理进度。当以上几个字段的和等于 total 字段时，请求就执行完毕了。

你可以使用 task id 查看某个任务。下例查看task id为 r1A2WoRbTwKZ516z6NEs5A：36619的任务信息：

  1. GET /_tasks/r1A2WoRbTwKZ516z6NEs5A:36619

该 API 可以与wait_for_comletion=false集成使用，可以清晰的查看已完成任务的状态。如果任务已经完成，并且在其上设置了wait_for_completion=false，那么请求将会返回结果或是错误字段。此功能的代价是当wait_for_completion=false时会在.tasks/task/${taskId}目录下会创建文档。您可以根据需要删除该文档。

6.4 取消任务(Works with the Cancel Task API)

任何_update_by_query操作都可以通过task cancel API来取消，如：

  1. POST _tasks/r1A2WoRbTwKZ516z6NEs5A:36619/_cancel

取消应该执行很快，但可能需要几秒钟。在此期间上面的 task status API将继续列出该任务，直到它完全被取消了。

6.5 阈值(Rethrottling)

在正在执行的请求中，requests_per_second的值可以在运行时通过_rethrotted API进行修改：

  1. POST _update_by_query/r1A2WoRbTwKZ516z6NEs5A:36619/_rethrottle?requests_per_second=-1

可以使用tasks API找到任务ID。

和 requests_per_seconds 参数设置一样，rethrottling 参数可以是 - 1 (禁用限制)或是其他十进制数(如 1.7 或 12 )。rethrottling 参数能提高查询速度且会立即生效，但是降低速度必须等到当前操作执行完后才起作用。这可以防止滚动超时

6.6 切片 (slicing)

update_by_query支持 sliced scroll 来使更新操作并行进行。这能提高效率并且提供了一种将请求分解为较小的部分的便捷方式。

6.6.1 手动切片 (Manually slicing)

通过为每个请求提供切片 ID 和切片总数，手动将 update_by_query操作进行分解：

  1. POST twitter/_update_by_query
  2. {
  3.   "slice": {
  4.     "id": 0,
  5.     "max": 2
  6.   },
  7.   "script": {
  8.     "source": "ctx._source['extra'] = 'test'"
  9.   }
  10. }
  11. POST twitter/_update_by_query
  12. {
  13.   "slice": {
  14.     "id": 1,
  15.     "max": 2
  16.   },
  17.   "script": {
  18.     "source": "ctx._source['extra'] = 'test'"
  19.   }
  20. }

你可以这样验证上述 api 的结果：

  1. GET _refresh
  2. POST twitter/_search?size=0&q=extra:test&filter_path=hits.total

返回如下的的结果：

  1. {
  2.   "hits": {
  3.     "total": 120
  4.   }
  5. }

6.6.2 自动切片 (Automatic slicing)

也可以让 update_by_query 自动并行地滚动切片。使用 slices 指定要使用的切片数：

  1. POST twitter/_update_by_query?refresh&slices=5
  2. {
  3.   "script": {
  4.     "source": "ctx._source['extra'] = 'test'"
  5.   }
  6. }

您可以通过下列语句验证运行结果：

  1. POST twitter/_search?size=0&q=extra:test&filter_path=hits.total

返回如下的的结果：

  1. {
  2.   "hits": {
  3.     "total": 120
  4.   }
  5. }

slices 设置为 auto 将允许 ElasticSearch 选择要使用的切片数。此设置将使用一个分片一个切片，直至达到某个限制。如果存在多个源索引，它将根据具有最少分片的那个索引所拥有的分片数来作为切片数。

向_delete_by_query 添加 slices 只会自动执行上一节中使用的手动过程，这意味着它有一些怪癖：

  • 您可以在 Tasks API 中查看这些请求。这些子请求是具有 slices 请求的任务的 “子 " 任务。
  • 仅使用 slices 获取请求的任务状态 (包含已完成切片的状态)。
  • 这些子请求可单独寻址，例如取消和重新限制。
  • 使用 slices 重新处理请求将按比例重新调整未完成的子请求。
  • 使用 slices 取消请求将取消每个子请求。
  • 由于 slices 的性质，每个子请求都不会获得完全均匀的文档部分。这些切片文档都会被处理，但某些切片可能比其他切片分到更大的文档。
  • 像 requests_per_second 这样的参数和带有 size 的 slices 请求 (按指定比例分配给每个子请求)。将其与上述关于分布不均匀的点相结合，您应该得出结论，使用 slices 的 size 可能不会删除指定大小的文档。
  • 每个子请求都会获得和源索引略有不同的快照，尽管这些快照几乎同时进行。

6.6.3 选择 slices 的数量

如果 slices 设置为 auto，elasticsearch 将会自动为大多数索引选择一个合理的数量。如果您设置手动切片或以其他方式来调整自动切片，请遵循以下准则：

当切片的数量等于索引的分片数时，查询性能最好。如果这个数量太大（如 500）, 请选择一个较小的数字，因为太多的切片会影响性能。设置高于分片数的切片通常不会提高效率反而会增加开销。

update 的性能与可用的切片数量呈正相关。

查询或更新是否是影响此时运行时性能的主要原因，这取决于reindexed时的文档和集群的资源。

6.7 获取新属性（pick up a new property）

假设您创建了一个没有动态映射的索引，然后向里面添加了数据，现在添加后一个 mapping value 可以从数据中获取更多的字段，如：

  1. PUT test
  2. {
  3.   "mappings": {
  4.     "_doc": {
  5.       "dynamic": false,   【1】
  6.       "properties": {
  7.         "text": {"type": "text"}
  8.       }
  9.     }
  10.   }
  11. }
  13. POST test/_doc?refresh
  14. {
  15.   "text": "words words",
  16.   "flag": "bar"
  17. }
  18. POST test/_doc?refresh
  19. {
  20.   "text": "words words",
  21.   "flag": "foo"
  22. }
  23. PUT test/_mapping/_doc   【2】
  24. {
  25.   "properties": {
  26.     "text": {"type": "text"},
  27.     "flag": {"type": "text", "analyzer": "keyword"}
  28.   }
  29. }

【1】"dynamic"：false。表示这个新字段不会被索引，只是存储到了_source。

【2】为了获取这个新的字段您必须reindex所有文档。

搜索数据将找不到任何内容：

  1. POST test/_search?filter_path=hits.total
  2. {
  3.   "query": {
  4.     "match": {
  5.       "flag": "foo"
  6.     }
  7.   }
  8. }
  1. {
  2.   "hits" : {
  3.     "total" : 0
  4.   }
  5. }

但是您可以用_update_by_query 请求获得新的 mapping：

  1. POST test/_update_by_query?refresh&conflicts=proceed
  2. POST test/_search?filter_path=hits.total
  3. {
  4.   "query": {
  5.     "match": {
  6.       "flag": "foo"
  7.     }
  8.   }
  9. }
  1. {
  2.   "hits" : {
  3.     "total" : 1
  4.   }
  5. }

添加多个字段时，也可以进行相同的操作。

原文链接：http://www.cnblogs.com/wtc1994/p/10707198.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Elasticsearch热门文章

  • ES 14 - (底层原理) Elasticsearch内部如何处理不同type的数据
  • ES 15 - Elasticsearch中的数据类型 (text、keyword、date、geo等)
  • java中elasticsearch7.x关于nested类型的api使用，新增+更新
  • Elasticsearch 映射参数详解 fields
  • [Elasticsearch] ES聚合场景下部分结果数据未返回问题分析
  • ES 22 - Elasticsearch中如何进行日期(数值)范围查询
  • Elasticsearch查询之Term?Query示例解析
  • ElasticSearch 实现分词全文检索 - SpringBoot 完整实现 Demo 附源码【完结篇】
  • ElasticStack学习（十）：深入ElasticSearch搜索之QueryFiltering、多/单字符串的多字段查询
  • ElasticSearch7 设置外网访问

    • Elasticsearch推荐文章

  • 利用Elasticsearch实现地理位置、城市搜索服务
  • 日志采集/分析
  • ELKF（elasticsearch、logstash、kibana、filebeat）搭建及收集nginx日志特昂糖-
  • ElasticSearch性能原理拆解
  • Elasticsearch使用实战以及代码详解
  • 使用 Docker Compose V2 快速搭建日志分析平台 ELK (Elasticsearch、Logstash 和 Kibana)
  • 【解决方案】MySQL5.7 百万数据迁移到 ElasticSearch7.x 的思考
  • K8s部署轻量级日志收集系统EFK（elasticsearch + filebeat + kibana）
  • Docker下elasticsearch8部署、扩容、基本操作实战(含kibana)
  • Elasticsearch 保姆级入门篇

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号