通過查詢 API 更新 · Elasticsearch 5.4 中文文檔

# 通過查詢 API 更新 _update_by_query?最簡單的用法只是基于索引來執行每個文檔的更新，而不更改源。這是非常有用快捷的，在完成一些 ?map 的變化也得以表現。這里是 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?`獲取索引的快照，當它開始和進行索引時發現使用的是?`internal?`的版本。這意味著如果在獲取快照時或者進行索引請求時文檔發生改變，那么將會發生版本沖突。當索引請求被處理的時間之間變化。當版本匹配文檔被更新時候，版本號遞增。注意：因為?`internal?`版本不支持的值 0 作為一個有效的版本號，與版本等于 0 時候，文檔無法使用?_update_by_query?更新，請求將會失敗。所有的更新和查詢失敗導致?`_update_by_query?`中止，并在返回錯誤的響應。正在執行的更新將會繼續下去。換句話說，該方法不支持回滾，僅中止。雖然第一次失敗將導致中止，由失敗的請求返回所有故障將在?`failures?`元素體現。因此，有可能存在一些失敗的實體。如果你想簡單地統計版本沖突不會導致?`_update_by_query`?中止你可以在 URL 中設置?`conflicts=proceed?`或在請求中設置?`"conflicts": "proceed"`?。第一個例子做到這一點因為它只是試圖獲取一個當前的 mapping 變化，這里發生的版本沖突僅僅意味著沖突的文檔在?`_update_by_query`?開始的時間和文件嘗試更新時間之間被更新了。這樣做具有一定的優點，因為更新將獲取當前 mapping 更新。回到API格式，您可以限制?`_update_by_query?`到一個單一類型。這將只從?`twitter?`索引中更新?`tweet 文件`： ``` POST twitter/tweet/_update_by_query?conflicts=proceed ``` 您也可以限制?`_update_by_query?`使用?。這將更新從?`twitter 索引中更新所有的文檔`： ``` POST twitter/_update_by_query?conflicts=proceed { "query": { （1） "term": { "user": "kimchy" } } } ``` （1）查詢必須作為一個值傳遞?`query?`鍵，就像在 Search API 中同樣的方式。還可以使用的?`q`?參數。、到目前為止，我們只是一直在更新文檔，而無需更改其來源。`_update_by_query?`支持腳本對象更新文檔。這將在所有 kimchy tweet 中增加?`likes` ： ``` POST twitter/_update_by_query { "script": { "inline": "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 不允許移動文件本身，只需修改其源。這是有意而為之，我們并沒有獲得從其原始位置刪除該文件的權限。通過多索引也可以完成所有的事情，就像搜索API： ``` POST twitter,blog/tweet,post/_update_by_query ``` 如果提供?`routing ，`則將被復制到滾動查詢，限制到分片的進程來匹配?`routing 值`： ``` POST twitter/_update_by_query?routing=1 ``` 在默認情況下?`_update_by_query?`使用 1000 批次的回滾。可以更改與批量大小的?`scroll_size URL`參數： ``` 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 ``` ## URL 參數除了標準的參數，如?`pretty`，通過查詢 API 還支持?`refresh`，`wait_for_completion`，`wait_for_active_shards?`和?`timeout`。當請求完成時，發送?`refresh?`將更新索引中的所有分片。這與索引 API 的?`refresh`?參數不同，這會使得僅僅分片獲取到最新的數據來作為索引。如果請求包含?`wait_for_completion=false 那么?`Elasticsearch 將執行一些預檢，啟動請求，然后返回一個?`task`?可與用于任務的 API 取消或獲取任務的狀態。Elasticsearch 還將創建此任務的記錄，在文檔?`.tasks/task/${taskId}中`。可以保留或刪除認為合適的。當你進行刪除時候，Elasticsearch 可以收回其使用的空間。 `wait_for_active_shards 控制著在請求時一個分片有多少個副本需要時活躍的`。`timeout?`控制每個寫請求的等待時間從不可用到可用。這兩個同時運行就像在 Bulk API 中一樣。 `requests_per_second?`可以被設置為任何正十進制數（`1.4，``6`，?`1000`等）并且禁用每秒字節流，或者它可以被設置為－1 來禁用字節流。禁用之后的等待時間，可以控制回滾延遲。等待時間不用于批處理完成時間和?`requests_per_second ＊ requests_in_the_batch 的時間。由于該批次并沒有分成多個批量傳輸的塊，大的數據塊將會導致?Elasticsearch 創建更多的請求導致較長時間的等待。這是突發的，不是平穩的。默認值為－1。` ## 響應 Json 響應如下： ``` { "took" : 639, "updated": 0, "batches": 1, "version_conflicts": 2, "retries": { "bulk": 0, "search": 0 } "throttled_millis": 0, "failures" : [ ] } ``` took ? ? ? ??從開始的毫秒數來結束整個操作過程。 updated ? ? ? ? 已成功更新的文件數量。 batches ? ? ? ??通過查詢請求返回響應的數目。 version_conflicts ? ? ? ? 版本沖突的次數。 retries ? ? ? ? 更新請求重試次數。bulk 是 buck action 重試的次數，search 是 search action 重試的次數。 throttled_millis ? ? ? ??符合?`requests_per_second 的毫秒數。` failures ? ? ? ? 所有索引失敗的集合。如果是非空的那么因為這些失敗請求失效。請參閱?`conflicts?`如何防止從版本沖突中停止運行。 ## Works With the Task API 你可以通過 Task API 獲取所有正在運行的更新請求狀態： ``` 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" : {（1） "total" : 6154, "updated" : 3500, "created" : 0, "deleted" : 0, "batches" : 4, "version_conflicts" : 0, "noops" : 0, "retries": { "bulk": 0, "search": 0 } "throttled_millis": 0 }, "description" : "" } } } } } ``` （1）這個對象的實際狀態。它就像響應中 json?用做 ?`total 的`重要補充。`total?`是操作任務的總數重新索引預計執行。您可以添加?`updated`，`created?`以及?`deleted 等多個`域。當它們之和等于?`total?`字段值時該請求將完成。通過 task id 可以直接獲得 task： ``` GET /_tasks/taskId:1 ``` 此 API 的優點是，它具有集成?`wait_for_completion=false`?到返回完成任務的狀態。如果任務完成，并?`wait_for_completion=false?`設置，將會返回?`results?`或?`error?`。此功能的成本，該文件?`wait_for_completion=false`創建的`.tasks/task/${taskId}`。它是由你來刪除該文檔。 ## Works With the Cancel Task API 通過 Task Cancle API 可以取消任何通過請求的更新： ``` POST _tasks / TASK_ID：1 / _cancel ``` 利用上述 task API 可以獲取到 task id。取消任務很快就會執行，通常需要幾秒鐘。上述的任務狀態 API 將任務以列表形式展示直到它被取消。 ## Rethrottling `requests_per_second 的值`可以在運行更新中可以通過使用?`_rethrottle?`的 API 來改變： ``` POST _update_by_query/task_id:1/_rethrottle?requests_per_second=-1 ``` 利用上述 task API 可以獲取到 task id。在設置上，它就像?`_update_by_query?`API 一樣，`requests_per_second`?可以設置成?`-1 來`禁用字節流和非整數。Rethrottling 是加快查詢需要，但是會減慢查詢在完成當前批后立即生效的效果，這可以防止回滾超時。 ## Manually slicing 通過請求的更新支持分片回滾，允許更加簡單的手動多線程操作： ``` POST twitter/_update_by_query { "slice": { "id": 0, "max": 2 }, "script": { "inline": "ctx._source['extra'] = 'test'" } } POST twitter/_update_by_query { "slice": { "id": 1, "max": 2 }, "script": { "inline": "ctx._source['extra'] = 'test'" } } ``` 你可以通過如下來驗證任務： ``` GET _refresh POST twitter/_search?size=0&q=extra:test&filter_path=hits.total ``` 結果 total 如下： ``` { "hits": { "total": 120 } } ``` ## 一個新的屬性假設你沒有創建動態映射索引，那么用數據填充它，然后添加一個映射值獲取數據字段： ``` {PUT test { "mappings": { "test": { "dynamic": false,（1） "properties": { "text": {"type": "text"} } } } } POST test/test?refresh { "text": "words words", "flag": "bar" } POST test/test?refresh { "text": "words words", "flag": "foo" } PUT test/_mapping/test（2） { "properties": { "text": {"type": "text"}, "flag": {"type": "text", "analyzer": "keyword"} } } ``` （1）這意味著，新的字段將不會被索引，只是存儲?`_source`。（2）這將更新映射并添加新的?`flag?`。添加了新的字段，你必須針對其重新索引。搜索數據將不會得到任何結果： ``` POST test/_search?filter_path=hits.total { "query": { "match": { "flag": "foo" } } } ``` ``` { "hits" : { "total" : 0 } } ``` 但是你可以發出一個?`_update_by_query?`請求，獲取新的映射： ``` POST test/_update_by_query?refresh&conflicts=proceed POST test/_search?filter_path=hits.total { "query": { "match": { "flag": "foo" } } } ``` ``` { "hits" : { "total" : 1 } } ``` 添加一個字段或者添加到多字段時，可以做同樣的事情。