如何提高ElasticSearch 索引速度
我Google了下，大致給出的答案如下：
- 使用bulk API
- 初次索引的時候，把 replica 設置為 0
- 增大 threadpool.index.queue_size
- 增大 indices.memory.index_buffer_size
- 增大 index.translog.flush_threshold_ops
- 增大 index.translog.sync_interval
- 增大 index.engine.robin.refresh_interval
這篇文章會講述上面幾個參數的原理，以及一些其他的思路。這些參數大體上是朝著兩個方向優化的：
- 減少磁盤寫入
- 增大構建索引處理資源
一般而言，通過第二種方式的需要慎用，會對集群查詢功能造成比較大的影響。
這里還有兩種形態的解決方案：
- 關閉一些特定場景并不需要的功能，比如Translog或者Version等
- 將部分計算挪到其他并行計算框架上，比如數據的分片計算等，都可以放到Spark上事先算好
上面的參數都和什么有關
- 其中 5,6 屬于 TransLog 相關。
- 4 則和Lucene相關
- 3 則因為ES里大量采用線程池，構建索引的時候，是有單獨的線程池做處理的
- 7 的話個人認為影響不大
- 2 的話，能夠使用上的場景有限。個人認為Replica這塊可以使用Kafka的ISR機制。所有數據還是都從Primary寫和讀。Replica盡量只作為備份數據。
Translog
為什么要有Translog? 因為Translog順序寫日志比構建索引更高效。我們不可能每加一條記錄就Commit一次，這樣會有大量的文件和磁盤IO產生。但是我們又想避免程序掛掉或者硬件故障而出現數據丟失，所以有了Translog，通常這種日志我們叫做Write Ahead Log。
為了保證數據的完整性，ES默認是每次request結束后都會進行一次sync操作。具體可以查看如下方法：
該方法會調用IndexShard.sync 方法進行文件落地。
你也可以通過設置index.translog.durability=async 來完成異步落地。這里的異步其實可能會有一點點誤導。前面是每次request結束后都會進行sync,這里的sync僅僅是將Translog落地。而無論你是否設置了async,都會執行如下操作：根據條件，主要是每隔sync_interval(5s) ，如果flush_threshold_ops(Integer.MAX_VALUE)，flush_threshold_size(512m),flush_threshold_period(30m) 滿足對應的條件，則進行flush操作，這里除了對Translog進行Commit以外，也對索引進行了Commit。
所以如果你是海量的日志，可以容忍發生故障時丟失一定的數據，那么完全可以設置，index.translog.durability=async，并且將前面提到的flush*相關的參數調大。
而極端情況，你還可以有兩個選擇：
- 設置index.translog.durability=async，接著設置index.translog.disable_flush=true進行禁用定時flush。然后你可以通過應用程序自己手動來控制flush。
- 通過改寫ES 去掉Translog日志相關的功能。
當然，如果去掉Translog日志有兩個風險點：
- Get***數據會有點問題。因為根據ID Get***數據是從Translog里拿的。
- 我們知道ES通過Shard Replication 保證Node節點出現故障時出現數據的完整性。在Relocating的時候，Replica 從Primary 進行Recover時，Primary會先Snapshot Lucene，然后拷貝數據到Replica,***通過回放Translog 保證數據的一致性。
Version
Version可以讓ES實現并發修改，但是帶來的性能影響也是極大的,這里主要有兩塊：
- 需要訪問索引里的版本號，觸發磁盤讀寫
- 鎖機制
目前而言，似乎沒有辦法直接關閉Version機制。你可以使用自增長ID并且在構建索引時，index 類型設置為create。這樣可以跳過版本檢查。
這個場景主要應用于不可變日志導入，隨著ES被越來越多的用來做日志分析，日志沒有主鍵ID,所以使用自增ID是合適的，并且不會進行更新，使用一個固定的版本號也是合適的。而不可變日志往往是追求吞吐量。
當然，如果有必要，我們也可以通過改寫ES相關代碼，禁用版本管理。
分發代理
ES是對索引進行了分片(Shard)，然后數據被分發到不同的Shard。這樣 查詢和構建索引其實都存在一個問題：
如果是構建索引，則需要對數據分揀，然后根據Shard分布分發到不同的Node節點上。
如果是查詢，則對外提供的Node需要收集各個Shard的數據做Merge
這都會對對外提供的節點造成較大的壓力，從而影響整個bulk/query 的速度。
一個可行的方案是，直接面向客戶提供構建索引和查詢API的Node節點都采用client模式，不存儲數據，可以達到一定的優化效果。
另外一個較為麻煩但似乎會更優的解決方案是，如果你使用類似Spark Streaming這種流式處理程序，在***往ES輸出的時候，可以做如下幾件事情：
- 獲取所有primary shard的信息，并且給所有shard帶上一個順序的數字序號，得到partition(順序序號) -> shardId的映射關系
- 對數據進行repartition,分區后每個partition對應一個shard的數據
- 遍歷這些partions,寫入ES。方法為直接通過RPC 方式，類似transportService.sendRequest 將數據批量發送到對應包含有對應ShardId的Node節點上。
這樣有三點好處：
- 所有的數據都被直接分到各個Node上直接處理。避免所有的數據先集中到一臺服務器
- 避免二次分發，減少一次網絡IO
- 防止***處理數據的Node壓力太大而導致木桶短板效應
場景
因為我正好要做日志分析類的應用，追求高吞吐量，這樣上面的三個優化其實都可以做了。一個典型只增不更新的日志入庫操作，可以采用如下方案：
- 對接Spark Streaming,在Spark里對數據做好分片，直接推送到ES的各個節點
- 禁止自動flush操作，每個batch 結束后手動flush。
- 避免使用Version
我們可以預期ES會產生多少個新的Segment文件，通過控制batch的周期和大小，預判出ES Segment索引文件的生成大小和Merge情況。***可能減少ES的一些額外消耗
總結
大體是下面這三個點讓es比原生的lucene吞吐量下降了不少：
為了數據完整性 ES額外添加了WAL(tanslog)
為了能夠并發修改 添加了版本機制
對外提供服務的node節點存在瓶頸
ES的線性擴展問題主要受限于第三點,
具體描述就是：
如果是構建索引，接受到請求的Node節點需要對數據分揀，然后根據Shard分布分發到不同的Node節點上。
如果是查詢，則對外提供的Node需要收集各個Shard的數據做Merge
另外，索引的讀寫并不需要向Master匯報。
