| 注册
请输入搜索内容

热门搜索

Java Linux MySQL PHP JavaScript Hibernate jQuery Nginx
ukon7587
4年前发布

自定义Spark Partitioner提升es-hadoop Bulk效率

   <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>    
 本文由用户 ukon7587 自行上传分享,仅供网友学习交流。所有权归原作者,若您的权利被侵害,请联系管理员。
 转载本站原创文章,请注明出处,并保留原始链接、图片水印。
 本站是一个以用户分享为主的开源技术平台,欢迎各类分享!
 本文地址:https://www.open-open.com/lib/view/open1460472838526.html
Spark Hadoop 分布式/云计算/大数据 ElasticSearch