5.document的批量操作及數據路由原理 · java核心知識整理

## 一、文檔批量操作 ### 1、批量查詢 ~~~ 1、批量查詢的好處就是一條一條的查詢，比如說要查詢100條數據，那么就要發送100次網絡請求，這個開銷還是很大的如果進行批量查詢的話，查詢100條數據，就只要發送1次網絡請求，網絡請求的性能開銷縮減100倍 2、mget的語法（1）一條一條的查詢 GET /test_index/test_type/1 GET /test_index/test_type/2 （2）mget批量查詢 GET /_mget { "docs" : [ { "_index" : "test_index", "_type" : "test_type", "_id" : 1 }, { "_index" : "test_index", "_type" : "test_type", "_id" : 2 } ] } { "docs": [ { "_index": "test_index", "_type": "test_type", "_id": "1", "_version": 2, "found": true, "_source": { "test_field1": "test field1", "test_field2": "test field2" } }, { "_index": "test_index", "_type": "test_type", "_id": "2", "_version": 1, "found": true, "_source": { "test_content": "my test" } } ] } （3）如果查詢的document是一個index下的不同type種的話 GET /test_index/_mget { "docs" : [ { "_type" : "test_type", "_id" : 1 }, { "_type" : "test_type", "_id" : 2 } ] } （4）如果查詢的數據都在同一個index下的同一個type下，最簡單了 GET /test_index/test_type/_mget { "ids": [1, 2] } 3、mget的重要性可以說mget是很重要的，一般來說，在進行查詢的時候，如果一次性要查詢多條數據的話，那么一定要用batch批量操作的api 盡可能減少網絡開銷次數，可能可以將性能提升數倍，甚至數十倍，非常非常之重要復制代碼 ~~~ ### 2、批量增刪改 ~~~ POST /_bulk { "delete": { "_index": "test_index", "_type": "test_type", "_id": "3" }} { "create": { "_index": "test_index", "_type": "test_type", "_id": "12" }} { "test_field": "test12" } { "index": { "_index": "test_index", "_type": "test_type", "_id": "2" }} { "test_field": "replaced test2" } { "update": { "_index": "test_index", "_type": "test_type", "_id": "1", "_retry_on_conflict" : 3} } { "doc" : {"test_field2" : "bulk test1"} } 每一個操作要兩個json串，語法如下： {"action": {"metadata"}} {"data"} 舉例，比如你現在要創建一個文檔，放bulk里面，看起來會是這樣子的： {"index": {"_index": "test_index", "_type", "test_type", "_id": "1"}} {"test_field1": "test1", "test_field2": "test2"} 有哪些類型的操作可以執行呢？（1）delete：刪除一個文檔，只要1個json串就可以了（2）create：PUT /index/type/id/_create，強制創建（3）index：普通的put操作，可以是創建文檔，也可以是全量替換文檔（4）update：執行的partial update操作 bulk api對json的語法，有嚴格的要求，每個json串不能換行，只能放一行，同時一個json串和一個json串之間，必須有一個換行 { "error": { "root_cause": [ { "type": "json_e_o_f_exception", "reason": "Unexpected end-of-input: expected close marker for Object (start marker at [Source: org.elasticsearch.transport.netty4.ByteBufStreamInput@5a5932cd; line: 1, column: 1])\n at [Source: org.elasticsearch.transport.netty4.ByteBufStreamInput@5a5932cd; line: 1, column: 3]" } ], "type": "json_e_o_f_exception", "reason": "Unexpected end-of-input: expected close marker for Object (start marker at [Source: org.elasticsearch.transport.netty4.ByteBufStreamInput@5a5932cd; line: 1, column: 1])\n at [Source: org.elasticsearch.transport.netty4.ByteBufStreamInput@5a5932cd; line: 1, column: 3]" }, "status": 500 } { "took": 41, "errors": true, "items": [ { "delete": { "found": true, "_index": "test_index", "_type": "test_type", "_id": "10", "_version": 3, "result": "deleted", "_shards": { "total": 2, "successful": 1, "failed": 0 }, "status": 200 } }, { "create": { "_index": "test_index", "_type": "test_type", "_id": "3", "_version": 1, "result": "created", "_shards": { "total": 2, "successful": 1, "failed": 0 }, "created": true, "status": 201 } }, { "create": { "_index": "test_index", "_type": "test_type", "_id": "2", "status": 409, "error": { "type": "version_conflict_engine_exception", "reason": "[test_type][2]: version conflict, document already exists (current version [1])", "index_uuid": "6m0G7yx7R1KECWWGnfH1sw", "shard": "2", "index": "test_index" } } }, { "index": { "_index": "test_index", "_type": "test_type", "_id": "4", "_version": 1, "result": "created", "_shards": { "total": 2, "successful": 1, "failed": 0 }, "created": true, "status": 201 } }, { "index": { "_index": "test_index", "_type": "test_type", "_id": "2", "_version": 2, "result": "updated", "_shards": { "total": 2, "successful": 1, "failed": 0 }, "created": false, "status": 200 } }, { "update": { "_index": "test_index", "_type": "test_type", "_id": "1", "_version": 3, "result": "updated", "_shards": { "total": 2, "successful": 1, "failed": 0 }, "status": 200 } } ] } bulk操作中，任意一個操作失敗，是不會影響其他的操作的，但是在返回結果里，會告訴你異常日志 POST /test_index/_bulk { "delete": { "_type": "test_type", "_id": "3" }} { "create": { "_type": "test_type", "_id": "12" }} { "test_field": "test12" } { "index": { "_type": "test_type" }} { "test_field": "auto-generate id test" } { "index": { "_type": "test_type", "_id": "2" }} { "test_field": "replaced test2" } { "update": { "_type": "test_type", "_id": "1", "_retry_on_conflict" : 3} } { "doc" : {"test_field2" : "bulk test1"} } POST /test_index/test_type/_bulk { "delete": { "_id": "3" }} { "create": { "_id": "12" }} { "test_field": "test12" } { "index": { }} { "test_field": "auto-generate id test" } { "index": { "_id": "2" }} { "test_field": "replaced test2" } { "update": { "_id": "1", "_retry_on_conflict" : 3} } { "doc" : {"test_field2" : "bulk test1"} } 2、bulk size最佳大小 bulk request會加載到內存里，如果太大的話，性能反而會下降，因此需要反復嘗試一個最佳的bulk size。一般從1000~5000條數據開始，嘗試逐漸增加。另外，如果看大小的話，最好是在5~15MB之間。復制代碼 ~~~ ## 二、數據路由 ### 1.document路由到shard上是什么意思？ ~~~ 一個index的數據會被分為多片，每片都在一個shard中，所以每個document只能存在一個shard中。當在客戶端創建document的時候，es就要決定這個document是要放在哪個index的shard上。這個過程就叫做document數據路由復制代碼 ~~~ ### 2.路由算法 ~~~ shard = hash(routing) % number_of_primary_shards 舉個例子，一個index有3個primary shard，P0，P1，P2 每次增刪改查一個document的時候，都會帶過來一個routing number，默認就是這個document的_id（可能是手動指定，也可能是自動生成） routing = _id，假設_id=1 會將這個routing值，傳入一個hash函數中，產出一個routing值的hash值，hash(routing) = 21 然后將hash函數產出的值對這個index的primary shard的數量求余數，21 % 3 = 0 就決定了，這個document就放在P0上。決定一個document在哪個shard上，最重要的一個值就是routing值，默認是_id，也可以手動指定，相同的routing值，每次過來，從hash函數中，產出的hash值一定是相同的無論hash值是幾，無論是什么數字，對number_of_primary_shards求余數，結果一定是在0~number_of_primary_shards-1之間這個范圍內的。0,1,2。復制代碼 ~~~ ### 3.\_id or custom routing value ~~~ 默認的routing就是_id 也可以在發送請求的時候，手動指定一個routing value，比如說put /index/type/id?routing=user_id 手動指定routing value是很有用的，可以保證說，某一類document一定被路由到一個shard上去，那么在后續進行應用級別的負載均衡，以及提升批量讀取的性能的時候，是很有幫助的復制代碼 ~~~ ### 4\. primary shard數量為什么不可變？ ~~~ 就是為了計算document 的路由。復制代碼 ~~~ ## 三、document內部增刪改原理 ~~~ （1）客戶端選擇一個node發送請求過去，這個node就是coordinating node（協調節點）（2）coordinating node，對document進行路由，將請求轉發給對應的node（有primary shard）（3）實際的node上的primary shard處理請求，然后將數據同步到replica node （4）coordinating node，如果發現primary node和所有replica node都搞定之后，就返回響應結果給客戶端復制代碼 ~~~ ![](data:image/svg+xml;utf8,) ## 四、document 讀請求內部原理（與上面大致差不多） ~~~ 1、客戶端發送請求到任意一個node，成為coordinate node 2、coordinate node對document進行路由，將請求轉發到對應的node，此時會使用round-robin隨機輪詢算法，在primary shard以及其所有replica中隨機選擇一個，讓讀請求負載均衡 3、接收請求的node返回document給coordinate node 4、coordinate node返回document給客戶端 5、特殊情況：document如果還在建立索引過程中，可能只有primary shard有，任何一個replica shard都沒有，此時可能會導致無法讀取到document，但是document完成索引建立之后，primary shard和replica shard就都有了復制代碼 ~~~ ## 五、寫一致性及 quorum機制 #### 1.寫一致性 ~~~ 如果有3個primary shard 必須primary shard都是活躍狀態才能執行寫入操作。復制代碼 ~~~ #### 2.quorum機制 ~~~ （1）consistency，one（primary shard），all（all shard），quorum（default）我們在發送任何一個增刪改操作的時候，比如說put /index/type/id，都可以帶上一個consistency參數，指明我們想要的寫一致性是什么？ put /index/type/id?consistency=quorum one：要求我們這個寫操作，只要有一個primary shard是active活躍可用的，就可以執行 all：要求我們這個寫操作，必須所有的primary shard和replica shard都是活躍的，才可以執行這個寫操作 quorum：默認的值，要求所有的shard中，必須是大部分的shard都是活躍的，可用的，才可以執行這個寫操作（2）quorum機制，寫之前必須確保大多數shard都可用，int( (primary + number_of_replicas) / 2 ) + 1，當number_of_replicas>1時才生效 quroum = int( (primary + number_of_replicas) / 2 ) + 1 舉個例子，3個primary shard，number_of_replicas=1，總共有3 + 3 * 1 = 6個shard quorum = int( (3 + 1) / 2 ) + 1 = 3 所以，要求6個shard中至少有3個shard是active狀態的，才可以執行這個寫操作（3）如果節點數少于quorum數量，可能導致quorum不齊全，進而導致無法執行任何寫操作 3個primary shard，replica=1，要求至少3個shard是active，3個shard按照之前學習的shard&replica機制，必須在不同的節點上，如果說只有1臺機器的話，是不是有可能出現說，3個shard都沒法分配齊全，此時就可能會出現寫操作無法執行的情況 1個primary shard，replica=3，quorum=((1 + 3) / 2) + 1 = 3，要求1個primary shard + 3個replica shard = 4個shard，其中必須有3個shard是要處于active狀態的。如果這個時候只有2臺機器的話，會出現什么情況呢？ es提供了一種特殊的處理場景，就是說當number_of_replicas>1時才生效，因為假如說，你就一個primary shard，replica=1，此時就2個shard (1 + 1 / 2) + 1 = 2，要求必須有2個shard是活躍的，但是可能就1個node，此時就1個shard是活躍的，如果你不特殊處理的話，導致我們的單節點集群就無法工作（4）quorum不齊全時，wait，默認1分鐘，timeout，100，30s 等待期間，期望活躍的shard數量可以增加，最后實在不行，就會timeout 我們其實可以在寫操作的時候，加一個timeout參數，比如說put /index/type/id?timeout=30，這個就是說自己去設定quorum不齊全的時候，es的timeout時長，可以縮短，也可以增長復制代碼 ~~~ ## 六、bulk api 奇特格式與底層性能優化 #### 1.前面批量增刪改大致提了下關于 bulk 的用法 ~~~ bulk api奇特的json格式 {"action": {"meta"}}\n {"data"}\n {"action": {"meta"}}\n {"data"}\n [{ "action": { }, "data": { } }] 1、bulk中的每個操作都可能要轉發到不同的node的shard去執行 2、如果采用比較良好的json數組格式允許任意的換行，整個可讀性非常棒，讀起來很爽，es拿到那種標準格式的json串以后，要按照下述流程去進行處理（1）將json數組解析為JSONArray對象，這個時候，整個數據，就會在內存中出現一份一模一樣的拷貝，一份數據是json文本，一份數據是JSONArray對象（2）解析json數組里的每個json，對每個請求中的document進行路由（3）為路由到同一個shard上的多個請求，創建一個請求數組（4）將這個請求數組序列化（5）將序列化后的請求數組發送到對應的節點上去 3、耗費更多內存，更多的jvm gc開銷我們之前提到過bulk size最佳大小的那個問題，一般建議說在幾千條那樣，然后大小在10MB左右，所以說，可怕的事情來了。假設說現在100個bulk請求發送到了一個節點上去，然后每個請求是10MB，100個請求，就是1000MB = 1GB，然后每個請求的json都copy一份為jsonarray對象，此時內存中的占用就會翻倍，就會占用2GB的內存，甚至還不止。因為弄成jsonarray之后，還可能會多搞一些其他的數據結構，2GB+的內存占用。占用更多的內存可能就會積壓其他請求的內存使用量，比如說最重要的搜索請求，分析請求，等等，此時就可能會導致其他請求的性能急速下降另外的話，占用內存更多，就會導致java虛擬機的垃圾回收次數更多，跟頻繁，每次要回收的垃圾對象更多，耗費的時間更多，導致es的java虛擬機停止工作線程的時間更多 4、現在的奇特格式 {"action": {"meta"}}\n {"data"}\n {"action": {"meta"}}\n {"data"}\n （1）不用將其轉換為json對象，不會出現內存中的相同數據的拷貝，直接按照換行符切割json （2）對每兩個一組的json，讀取meta，進行document路由（3）直接將對應的json發送到node上去 5、最大的優勢在于，不需要將json數組解析為一個JSONArray對象，形成一份大數據的拷貝，浪費內存空間，盡可能地保證性能 ~~~ 作者：Leo\_CX330 鏈接：https://juejin.cn/post/6926344149796913166 來源：掘金著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。