前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >ES-Spark连接ES后,ES Client节点流量打满分析

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

作者头像
YG
发布2018-05-23 17:17:54
3.2K3
发布2018-05-23 17:17:54
举报
文章被收录于专栏:YG小书屋

问题描述

前段时间用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:

代码语言:javascript
复制
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:

代码语言:javascript
复制
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):

代码语言:javascript
复制
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中,他会打乱所有的数据节点,并从中选出一个节点用来通信,如下:

代码语言:javascript
复制
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中随机选择一个进行写操作。

代码语言:javascript
复制
if (settings.getNodesWANOnly()) {
  return randomNodeWrite(settings, currentInstance, resource, log);
}

以bulk为例,其操作过程如下:

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

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

代码语言:javascript
复制
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)。

代码语言:javascript
复制
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提高并发度。

代码语言:javascript
复制
if (version.onOrAfter(EsMajorVersion.V_5_X)) {
  partitions = findSlicePartitions(client.getRestClient(), settings, mapping, nodesMap, shards);
} else {
  partitions = findShardPartitions(settings, mapping, nodesMap, shards);
}

在findSlicePartitions中给出了计算公式:

代码语言:javascript
复制
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)));
}
本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2018.02.08 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 问题描述
  • 解决方法
  • 原因分析
  • 源码角度分析
    • 1、es-spark 读
      • 2、es-spark 写
        • 3、shard-partition 对应关系
        相关产品与服务
        Elasticsearch Service
        腾讯云 Elasticsearch Service(ES)是云端全托管海量数据检索分析服务,拥有高性能自研内核,集成X-Pack。ES 支持通过自治索引、存算分离、集群巡检等特性轻松管理集群,也支持免运维、自动弹性、按需使用的 Serverless 模式。使用 ES 您可以高效构建信息检索、日志分析、运维监控等服务,它独特的向量检索还可助您构建基于语义、图像的AI深度应用。
        领券
        问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档