<h2>前言</h2>    <p>之前写过一篇文章， <a href="/misc/goto?guid=4959669452361023073" rel="nofollow,noindex">如何提高ElasticSearch 索引速度</a> 。除了对ES本身的优化以外，我现在大体思路是尽量将逻辑外移到Spark上,Spark的分布式计算能力强，cpu密集型的很适合。这篇文章涉及的调整也是对 <a href="/misc/goto?guid=4959669052362094274" rel="nofollow,noindex">SparkES 多维分析引擎设计</a> 中提及的一个重要概念“shard to partition ,partition to shard ” 的实现。不过目前只涉及到构建索引那块。</p>    <h2>问题描述</h2>    <p>当你bulk数据到集群，按照 <a href="/misc/goto?guid=4959670386912947956" rel="nofollow,noindex">ElasticSearch Bulk 源码解析</a> 所描述的:</p>    <p>接着通过executeBulk方法进入原来的流程。在该方法中，对bulkRequest.requests 进行了两次for循环。</p>    <p>第一次判定如果是IndexRequest就调用IndexRequest.process方法，主要是为了解析出timestamp,routing,id,parent 等字段。</p>    <p>第二次是为了对数据进行分拣。大致是为了形成这么一种结构：</p>    <p>第二次就是对提交的数据进行分拣，然后根据route/_id 等值找到每个数据所属的Shard，最后将数据发送到对应Shard所在的Node节点上。</p>    <p>然而这导致了两个问题：</p>    <ol>     <li>ES Node之间会形成N*N个连接，消耗掉过多的bulk线程</li>     <li>出现了很多并不需要的网络IO</li>    </ol>    <p>所以我们希望能够避免这种情况。</p>    <h2>Spark Partition to ES Shard</h2>    <p>我们希望能够将分拣的逻辑放到Spark端，保证Spark 的Partition 和ES的Shard 一一对应，并且实现特定的Partitoner 保证数据到达ES都会被对应的Shard所在的节点直接消费，而不会再被转发到其他节点。经过我的实际测试，做了该调整后，写入QPS有两倍以上的提升</p>    <h2>理论基础</h2>    <p>这里的理论基础自然是es-hadoop项目。</p>    <p>类的调用路径关系为：</p>    <pre>  <code class="language-java">EsSpark ->        EsRDDWriter ->              RestService ->                     RestRepository ->                               RestClient ->                                  NetworkClient ->                                           CommonsHttpTransport</code></pre>    <p>简单介绍下他们的作用：</p>    <ul>     <li>EsSpark, 读取ES和存储ES的入口。通过隐式转换，会显得更Spark.</li>     <li>EsRDDWriter ,调用RestService创建PartitionWriter,对ES进行数据写入</li>     <li>RestService，负责创建 RestRepository，PartitionWriter</li>     <li>RestRepository，bulk高层抽象，底层利用NetworkClient做真实的http请求，另外也维护Buffer相关的，典型比如积攒了多少条，多少M之后进行flush等。</li>     <li>NetworkClient 对 CommonsHttpTransport的封装，主要添加了一些节点校验功能。</li>     <li>CommonsHttpTransport 你可以认为是对HttpClient的一个封装</li>    </ul>    <p>原来我以为需要对es-hadoop项目的源码进行修改才能实现前面提到的逻辑。事实上基于es-hadoop很容易实现上面提到的需求。</p>    <p>在RestService类里，构建RestRepository的时候，会判定是多索引还是单索引。对应代码如下：</p>    <pre>  <code class="language-java">RestRepository repository = (iformat.hasPattern() ?   initMultiIndices(settings, currentSplit, resource, log) :   initSingleIndex(settings, currentSplit, resource, log));</code></pre>    <p>这里我们只解析单索引部分代码，在对应的initSingleIndex方法里有如下代码：</p>    <pre>  <code class="language-java">int bucket = currentInstance % targetShards.size();  Shard chosenShard = orderedShards.get(bucket);  Node targetNode = targetShards.get(chosenShard);</code></pre>    <ul>     <li>targetShards 是索引所有的主分片到对应Node节点的映射。</li>     <li>orderedShards 则是根据shardId 顺序排序Shard集合</li>     <li>currentInstance 是partitionId</li>    </ul>    <p>这里我们假设partitonId 和 shardId 是对应的。也就是partitionId 0 里的数据，都是属于shardId 为0,则，</p>    <ul>     <li>currentInstance < targetShards.size()</li>     <li>bucket == currentInstance</li>     <li>对应的targetNode持有的Shard正好可以消化掉currentInstance partition 分区的数据</li>    </ul>    <p>所以我们唯一要做的就是实现一个Partitioner就好。</p>    <h2>ESShardPartitioner 实现</h2>    <p>涉及到这块的主要有 es-hadoop 的mr以及 spark模块。在mr模块里包含了ES的分片规则实现。 spark 模块则包含ESShardPartitioner类。</p>    <p>代码如下：</p>    <pre>  <code class="language-java">package org.elasticsearch.spark  import ....  class ESShardPartitioner(settings:String) extends Partitioner {        protected val log = LogFactory.getLog(this.getClass())          protected var _numPartitions = -1           override def numPartitions: Int = {             val newSettings = new PropertiesSettings().load(settings)          val repository = new RestRepository(newSettings)          val targetShards = repository.getWriteTargetPrimaryShards(newSettings.getNodesClientOnly())          repository.close()          _numPartitions = targetShards.size()          _numPartitions        }          override def getPartition(key: Any): Int = {          val shardId = ShardAlg.shard(key.toString(), _numPartitions)          shardId        }  }    public class ShardAlg {      public static int shard(String id, int shardNum) {          int hash = Murmur3HashFunction.hash(id);          return mod(hash, shardNum);      }        public static int mod(int v, int m) {          int r = v % m;          if (r < 0) {              r += m;          }          return r;      }  }</code></pre>    <p>使用方式如下：</p>    <pre>  <code class="language-java">......partitionBy(new ESShardPartitioner(settings)).foreachPartition { iter =>        try {          val newSettings = new PropertiesSettings().load(settings)          //创建EsRDDWriter          val writer = EsRDDCreator.createWriter(newSettings.save())          writer.write(TaskContext.get(), iter.map(f => f._2))                }</code></pre>    <p>不过这种方式也是有一定大问题，就是如果ES的Shard数目比较小，导致Spark的partition数也小，写入并发性会受到影响。</p>    <p>所以这里有第二套方案：</p>    <ol>     <li>修改ESShardPartitioner，可以让多个分区对应一个Shard</li>     <li>每个分区通过EsRDDWriter指定shardId进行写入。</li>    </ol>    <p>第二点可能需要修改es-hadoop源码了，不过修改也很简单，通过settings传递shardId,然后在RestService.initSingleIndex添加如下代码：</p>    <pre>  <code class="language-java">if(settings.getProperty(ConfigurationOptions.ES_BULK_SHARDID) != null){     targetNode = targetShards.get(  Integer.parseInt(  settings.getProperty(ConfigurationOptions.ES_BULK_SHARDID)));  }</code></pre>    <p>在创建EsRDDWriter时拷贝settings的副本并且加入对应的ConfigurationOptions.ES_BULK_SHARDID.</p>    <p>来源：http://www.jianshu.com/p/cccc56e39429?utm_source=tuicool&utm_medium=referral</p>