ES-Spark连接ES后，ES Client节点流量打满分析

问题描述

前段时间用es-spark读取es数遇到了client节点流量打满的现象。es-spark配置的es.nodes是es的域名。由于其中一个client是master节点，然后普通查询变得特别慢，运行20多分钟后，主节点崩溃。

解决方法

临时解决方案：降低es-spark的并发，并重启主节点。

最终解决方案：设置es.nodes.wan.only为false，即不用域名访问。将es.nodes配置为client节点的IP。

原因分析

域名访问时必须配置参数es.nodes.wan.only为true，关于该参数的解释如下：

Whether the connector is used against an Elasticsearch instance in a cloud/restricted environment over the WAN, such as Amazon Web Services. In this mode, the connector disables discovery and onlyconnects through the declared es.nodes during all operations, including reads and writes. Note that in this mode, performance is highly affected.

es.nodes.wan.only设置为true时即只通过client节点进行读取操作，因此主节点负载会特别高，性能很差。长时间运行后，java gc回收一次要几十秒，慢慢的OOM，系统崩溃。

配置es.nodes为client节点的IP后，spark只通过data节点访问ES：

es.nodes.data.only (default true)
Whether to use Elasticsearch data nodes only. When enabled, elasticsearch-hadoop will route all its requests (after nodes discovery, if enabled) through the data nodes within the cluster. The purpose of this configuration setting is to avoid overwhelming non-data nodes as these tend to be "smaller" nodes. This is enabled by default.

es.nodes.data.only 默认为true，即spark所有的请求都会发到数据节点，不在通过client节点进行请求的转发，client节点只用来服务普通的查询。

源码角度分析

1、es-spark 读

其架构图如下所示：

es_spark_read.png

我们知道spark能动态的发现节点,，但当我们配置wan.only为true的时候，整个集群的节点IP中只有从域名中解析出来的IP：

private static List<String> qualifyNodes(String nodes, int defaultPort, boolean resolveHostNames)
 {
   List<String> list = StringUtils.tokenize(nodes);
   for (int i = 0; i < list.size(); i++)
   {
     String nodeIp = resolveHostNames ? resolveHostToIpIfNecessary((String)list.get(i)) : (String)list.get(i);
     list.set(i, qualifyNode(nodeIp, defaultPort));
   }
   return list;
 }

从源码角度以scroll为例：

JavaEsSpark.esJsonRDD()-->JavaEsRDD.compute()-->JavaEsRDDIterator(继承AbstractEsRDDIterator).reader$lzycompute()
在lzycompute方法中我们可以看到，执行请求的是RestService：

private ScrollQuery reader$lzycompute()
 {
   synchronized (this)
   {
     if (!this.bitmap$0)
     {
       initialized_$eq(true);
       Settings settings = this.partition.settings();

       initReader(settings, log());

       RestService.PartitionReader readr = RestService.createReader(settings, this.partition, log());this.reader =
         readr.scrollQuery();this.bitmap$0 = true;
     }
     return this.reader;
   }
 }

在createReader方法中会判断spark节点和当前请求请求的shard是否是同一个节点，如果是同一个节点，则将该IP写入Setting，用本地节点IP进行请求（执行请求的时候，从setting中读取该ip）：

if ((!SettingsUtils.hasPinnedNode(settings)) && (partition.getLocations().length > 0))
{
  String pinAddress = checkLocality(partition.getLocations(), log);
  if (pinAddress != null)
  {
    if (log.isDebugEnabled()) {
      log.debug(String.format("Partition reader instance [%s] assigned to [%s]:[%s]", new Object[] { partition, pinAddress }));
    }
    SettingsUtils.pinNode(settings, pinAddress);
  }
}

通过PartitionReader.scrollQuery()-->SearchRequestBuilder.build()-->RestRepository.scanLimit()-->ScrollQuery.hasNext()-->RestRepository.scroll()-->RestClient.execute()-->NetWorkClient.execute()-->Transport.execute()

其实我们看到的最终要的执行是在NetWorkClient中，他会打乱所有的数据节点，并从中选出一个节点用来通信，如下：

public NetworkClient(Settings settings, TransportFactory transportFactory)
{
  this.settings = settings.copy();
  this.nodes = SettingsUtils.discoveredOrDeclaredNodes(settings);
  this.transportFactory = transportFactory;

  Collections.shuffle(this.nodes);//打乱排序
  if (SettingsUtils.hasPinnedNode(settings))
  {
    String pinnedNode = SettingsUtils.getPinnedNode(settings);
    if (log.isDebugEnabled()) {
      log.debug("Opening (pinned) network client to " + pinnedNode);
    }
    this.nodes.remove(pinnedNode);
    this.nodes.add(0, pinnedNode);
  }
  selectNextNode();

  Assert.notNull(this.currentTransport, "no node information provided");
}


private boolean selectNextNode()
{
  if (this.nextClient >= this.nodes.size()) {
    return false;
  }
  if (this.currentTransport != null) {
    this.stats.nodeRetries += 1;
  }
  closeTransport();
  this.currentNode = ((String)this.nodes.get(this.nextClient++));
  SettingsUtils.pinNode(this.settings, this.currentNode);
  this.currentTransport = this.transportFactory.create(this.settings, this.currentNode);
  return true;
}

2、es-spark 写

其架构图如下所示：

es_spark_write.png

从源码角度来看：
写请求的时候，如果wan.only配置为true，则节点IP就是从域名解析出的IP中随机选择一个进行写操作。

if (settings.getNodesWANOnly()) {
  return randomNodeWrite(settings, currentInstance, resource, log);
}

以bulk为例，其操作过程如下：

EsSpark.doSaveToEs()-->EsRDDWriter.write()-->RestService.createWriter()

在createWriter中首先随机或者按照split选择一个节点：

int selectedNode = currentSplit < 0 ? new Random().nextInt(nodes.size()) : currentSplit % nodes.size();
SettingsUtils.pinNode(settings, (String)nodes.get(selectedNode));

最终的改变是在RestService的initSingleIndex方法中，通过根据当前的split，找到对应的shard，然后获取到shard所在的IP，写入setting中（执行请求的时候，从setting中读取该ip）。

if (currentInstance <= 0) {
   currentInstance = new Random().nextInt(targetShards.size()) + 1;
 }
 int bucket = currentInstance % targetShards.size();
 ShardInfo chosenShard = (ShardInfo)orderedShards.get(bucket);
 NodeInfo targetNode = (NodeInfo)targetShards.get(chosenShard);

 SettingsUtils.pinNode(settings, targetNode.getPublishAddress());
 String node = SettingsUtils.getPinnedNode(settings);
 repository = new RestRepository(settings);

接下来就是RestRepository.tryFlush()-->RestClient.bulk()-->NetWorkClient.execute()-->Transport.execute()，这一套流程和读差不多，这里就不再介绍。

3、shard-partition 对应关系

es-spark写的话就是就是一个partition对应一个shard，这里从上述的es-spark写代码中可以看出，不再过多介绍。

es-spark读的时候是按照shard的文档数来分的:

partition=numberOfDoc(shard)/100000

100000是默认的配置，这个可通过es.input.max.docs.per.partition配置。

假设一个shard有23w条doc，10w条一个partition，则分为3个partition。读操作时shard-partition 的架构图如下所示：

partition_shard.png

从源码角度来说，如果是5.X版本，则用scrollSlice提高并发度。

if (version.onOrAfter(EsMajorVersion.V_5_X)) {
  partitions = findSlicePartitions(client.getRestClient(), settings, mapping, nodesMap, shards);
} else {
  partitions = findShardPartitions(settings, mapping, nodesMap, shards);
}

在findSlicePartitions中给出了计算公式：

for (List<Map<String, Object>> group : shards)
{
  String index = null;
  int shardId = -1;
  List<String> locationList = new ArrayList();
  for (Map<String, Object> replica : group)
  {
    ShardInfo shard = new ShardInfo(replica);
    index = shard.getIndex();
    shardId = shard.getName().intValue();
    if (nodes.containsKey(shard.getNode())) {
      locationList.add(((NodeInfo)nodes.get(shard.getNode())).getPublishAddress());
    }
  }
  String[] locations = (String[])locationList.toArray(new String[0]);
  StringBuilder indexAndType = new StringBuilder(index);
  if (StringUtils.hasLength(types))
  {
    indexAndType.append("/");
    indexAndType.append(types);
  }
  long numDocs = client.count(indexAndType.toString(), Integer.toString(shardId), query);
  int numPartitions = (int)Math.max(1L, numDocs / maxDocsPerPartition);
  for (int i = 0; i < numPartitions; i++)
  {
    PartitionDefinition.Slice slice = new PartitionDefinition.Slice(i, numPartitions);
    partitions.add(new PartitionDefinition(settings, mapping, index, shardId, slice, locations));
  }
}

public int getMaxDocsPerPartition()
{
  return Integer.parseInt(getProperty("es.input.max.docs.per.partition", Integer.toString(100000)));
}