如何提高ElasticSearch 索引速度

szwangxz 3年前

来自: http://www.jianshu.com/p/5eeeeb4375d4

我Google了下,大致给出的答案如下:

  1. 使用bulk API
  2. 初次索引的时候,把 replica 设置为 0
  3. 增大 threadpool.index.queue_size
  4. 增大 indices.memory.index_buffer_size
  5. 增大 index.translog.flush_threshold_ops
  6. 增大 index.translog.sync_interval
  7. 增大 index.engine.robin.refresh_interval

其中 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操作。具体可以查看如下方法:

org.elasticsearch.action.bulk.TransportShardBulkAction.processAfter

该方法会调用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*相关的参数调大。

而极端情况,你还可以有两个选择:

  1. 设置 index.translog.durability=async ,接着设置 index.translog.disable_flush=true 进行禁用定时flush。然后你可以通过应用程序自己手动来控制flush。

  2. 通过改写ES 去掉Translog日志相关的功能

Version

Version可以让ES实现并发修改,但是带来的性能影响也是极大的,这里主要有两块:

  1. 需要访问索引里的版本号,触发磁盘读写
  2. 锁机制

目前而言,似乎没有办法直接关闭Version机制。你可以使用自增长ID并且在构建索引时,index 类型设置为create。这样可以跳过版本检查。

这个场景主要应用于不可变日志导入,随着ES被越来越多的用来做日志分析,日志没有主键ID,所以使用自增ID是合适的,并且不会进行更新,使用一个固定的版本号也是合适的。而不可变日志往往是追求吞吐量。

当然,如果有必要,我们也可以通过改写ES相关代码,禁用版本管理。

分发代理

ES是对索引进行了分片(Shard),然后数据被分发到不同的Shard。这样 查询和构建索引其实都存在一个问题:

如果是构建索引,则需要对数据分拣,然后根据Shard分布分发到不同的Node节点上。如果是查询,则对外提供的Node需要收集各个Shard的数据做Merge

这都会对对外提供的节点造成较大的压力。 解决这个有效的方式,如果你使用类似Spark Streaming这种流式处理程序,在最后往ES输出的时候,根据将所有的数据实现按照shard进行分区(Partition),然后得到 partition -> shardId 的映射关系,直接通过RPC 方式,类似 transportService.sendRequest 将数据批量发送到发送到对应包含有对应ShardId的Node节点上。

这样有三点好处:

  1. 所有的数据都被直接分到各个Node上直接处理。避免所有的数据先集中到一台服务器
  2. 避免二次分发,减少一次网络IO
  3. 防止最先处理数据的Node压力太大而导致木桶短板效应

当然,还有一种办法是,直接面向客户提供构建索引和查询API的Node节点都采用client模式,不存储数据,可以达到一定的优化效果。

一些题外话

其实如果对接Spark Streaming时,完全可以禁止自动flush操作,每个batch 手动提交,这样,一个batch周期,对应可能产生可预期的Segment文件,通过控制batch的周期和大小,甚至可以预判出ES Segment索引文件的生成大小和Merge情况。

场景

因为我正好要做日志分析类的应用,追求高吞吐量,这样上面的三个优化其实都可以做了。

</div>