6、Update By Query API
_update_by_query
接口可以在不改变 source 的情况下对 index 中的每个文档进行更新。这对于获取新属性或其他联机映射更改很有用。以下是 API:
POST twitter/_update_by_query?conflicts=proceed
这将返回如下内容:
{
"took" : 147,
"timed_out": false,
"updated": 120,
"deleted": 0,
"batches": 1,
"version_conflicts": 0,
"noops": 0,
"retries": {
"bulk": 0,
"search": 0
},
"throttled_millis": 0,
"requests_per_second": -1.0,
"throttled_until_millis": 0,
"total": 120,
"failures" : [ ]
}
_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:
POST twitter/_doc/_update_by_query?conflicts=proceed
您还可以使用 Query DSL 限制 _update_by_query,下面的例子将会更新 twitter 索引中所有 user 字段为 kimchy 的文档:
POST twitter/_update_by_query?conflicts=proceed
{
"query": {
"term": {
"user": "kimchy"
}
}
}
目前,我们在更新文档的时候没有更改 source。这对诸如获取新属性很有用,但只是其中一般的乐趣。_update_by_query 支持用脚本来更新文档。下面的例子将会把所有user为 kimchy 的 likes 字段加 1:
POST twitter/_update_by_query
{
"script": {
"source": "ctx._source.likes++",
"lang": "painless"
},
"query": {
"term": {
"user": "kimchy"
}
}
}
正如在 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也可以一次修改多个索引和多个类型,如:
POST twitter,blog/_doc,post/_update_by_query
如何设置了routing
,那么操作的分片还应该满足路由的条件,如:
POST twitter/_update_by_query?routing=1
_update_by_query 默认使用的scroll 批次大小是 1000,您可以通过设置 scroll_size
来 修改此值:
POST twitter/_update_by_query?scroll_size=100
_update_by_query 也可以指定 Ingest Node接口的 pipeline,如:
PUT _ingest/pipeline/set-foo
{
"description" : "sets foo",
"processors" : [ {
"set" : {
"field": "foo",
"value": "bar"
}
} ]
}
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:
target_time = 1000 / 500 per second = 2 seconds
wait_time = target_time - write_time = 2 seconds - .5 seconds = 1.5 seconds
由于该批处理是作为单个 _bulk 请求发出的,因此大批量的请求将导致ElasticSearch创建多个请求,然后在启动下一个集合之前等待一段时间。这是“突发”而不是“平稳”。默认值为-1。
6.2 响应体(Response body)
{
"took" : 147,
"timed_out": false,
"total": 5,
"updated": 5,
"deleted": 0,
"batches": 1,
"version_conflicts": 0,
"noops": 0,
"retries": {
"bulk": 0,
"search": 0
},
"throttled_millis": 0,
"requests_per_second": -1.0,
"throttled_until_millis": 0,
"failures" : [ ]
}
- 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
- 执行失败的数组,包含在执行过程中任何不可恢复的错误。如果这个数组不是空的,那么请求会因为这些失败而中止。_updata_by_query是使用批处理实现的,任何失败都会导致整个执行被中止。可以使用conflicts参数来防止reindex在版本冲突时造成操作中止。
6.3 结合 taskAPi 使用(Works with the Task API)
您可以使用 Task API 获取任何正在进行 update_by_query 请求的状态:
GET _tasks?detailed=true&actions=*byquery
返回值:
{
"nodes" : {
"r1A2WoRbTwKZ516z6NEs5A" : {
"name" : "r1A2WoR",
"transport_address" : "127.0.0.1:9300",
"host" : "127.0.0.1",
"ip" : "127.0.0.1:9300",
"attributes" : {
"testattr" : "test",
"portsfile" : "true"
},
"tasks" : {
"r1A2WoRbTwKZ516z6NEs5A:36619" : {
"node" : "r1A2WoRbTwKZ516z6NEs5A",
"id" : 36619,
"type" : "transport",
"action" : "indices:data/write/update/byquery",
"status" : {
"total" : 6154,
"updated" : 3500,
"created" : 0,
"deleted" : 0,
"batches" : 4,
"version_conflicts" : 0,
"noops" : 0,
"retries": {
"bulk": 0,
"search": 0
},
"throttled_millis": 0
},
"description" : ""
}
}
}
}
}
status:这个对象包含了当前任务的实际状态。total
字段是本次操作需要重新索引的文档数。你可以通过 updated
, created
, and deleted
字段估计处理进度。当以上几个字段的和等于 total
字段时,请求就执行完毕了。
你可以使用 task id 查看某个任务。下例查看task id为 r1A2WoRbTwKZ516z6NEs5A:36619
的任务信息:
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来取消,如:
POST _tasks/r1A2WoRbTwKZ516z6NEs5A:36619/_cancel
取消应该执行很快,但可能需要几秒钟。在此期间上面的 task status API将继续列出该任务,直到它完全被取消了。
6.5 阈值(Rethrottling)
在正在执行的请求中,requests_per_second
的值可以在运行时通过_rethrotted
API进行修改:
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操作进行分解:
POST twitter/_update_by_query
{
"slice": {
"id": 0,
"max": 2
},
"script": {
"source": "ctx._source['extra'] = 'test'"
}
}
POST twitter/_update_by_query
{
"slice": {
"id": 1,
"max": 2
},
"script": {
"source": "ctx._source['extra'] = 'test'"
}
}
你可以这样验证上述 api 的结果:
GET _refresh
POST twitter/_search?size=0&q=extra:test&filter_path=hits.total
返回如下的的结果:
{
"hits": {
"total": 120
}
}
6.6.2 自动切片 (Automatic slicing)
也可以让 update_by_query 自动并行地滚动切片。使用 slices
指定要使用的切片数:
POST twitter/_update_by_query?refresh&slices=5
{
"script": {
"source": "ctx._source['extra'] = 'test'"
}
}
您可以通过下列语句验证运行结果:
POST twitter/_search?size=0&q=extra:test&filter_path=hits.total
返回如下的的结果:
{
"hits": {
"total": 120
}
}
slices 设置为 auto 将允许 ElasticSearch 选择要使用的切片数。此设置将使用一个分片一个切片,直至达到某个限制。如果存在多个源索引,它将根据具有最少分片的那个索引所拥有的分片数来作为切片数。
向_updata_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 可以从数据中获取更多的字段,如:
PUT test
{
"mappings": {
"_doc": {
"dynamic": false, 【1】
"properties": {
"text": {"type": "text"}
}
}
}
}
POST test/_doc?refresh
{
"text": "words words",
"flag": "bar"
}
POST test/_doc?refresh
{
"text": "words words",
"flag": "foo"
}
PUT test/_mapping/_doc 【2】
{
"properties": {
"text": {"type": "text"},
"flag": {"type": "text", "analyzer": "keyword"}
}
}
【1】"dynamic":false。表示这个新字段不会被索引,只是存储到了_source。
【2】为了获取这个新的字段您必须reindex所有文档。
搜索数据将找不到任何内容:
POST test/_search?filter_path=hits.total
{
"query": {
"match": {
"flag": "foo"
}
}
}
{
"hits" : {
"total" : 0
}
}
但是您可以用_update_by_query 请求获得新的 mapping:
POST test/_update_by_query?refresh&conflicts=proceed
POST test/_search?filter_path=hits.total
{
"query": {
"match": {
"flag": "foo"
}
}
}
{
"hits" : {
"total" : 1
}
}
添加多个字段时,也可以进行相同的操作。