StarRocks 中关于 Hadoop Hedged Read 性能测试

Hedged Read 简介

If a read from a block is slow, start up another parallel, ‘hedged’ read against a different block replica. We then take the result of which ever read returns first (the outstanding read is cancelled). This ‘hedged’ read feature will help rein in the outliers, the odd read that takes a long time because it hit a bad patch on the disc, etc.

This feature is off by default. To enable this feature, set dfs.client.hedged.read.threadpool.size to a positive number. The threadpool size is how many threads to dedicate to the running of these ‘hedged’, concurrent reads in your client.

Then set dfs.client.hedged.read.threshold.millis to the number of milliseconds to wait before starting up a ‘hedged’ read. For example, if you set this property to 10, then if a read has not returned within 10 milliseconds, we will start up a new read against a different block replica.

This feature emits new metrics（几个新指标参数介绍）:

• hedgedReadOps – 发起 hedged read 操作的次数
• hedgeReadOpsWin – how many times the hedged read ‘beat’ the original read
• hedgedReadOpsInCurThread – how many times we went to do a hedged read but we had to run it in the current thread because dfs.client.hedged.read.threadpool.size was at a maximum.

hdfs-site.xml 配置参数模版

<configuration> 
  <property>
     <name>dfs.client.hedged.read.threadpool.size</name>
     <value>128</value>
   </property>
   <property>
     <name>dfs.client.hedged.read.threshold.millis</name>
     <value>2000</value>
   </property>
</configuration>  

测试环境

就跑 100G 的 TPCH Parquet 测试，数据存储在 emr 的 hdfs 上面，2 副本，以此避免数据全部都短路读，不会涉及 network io。（EMR 有 3 个 worker 节点）。

StarRocks 配置：1 FE + 2 BE

通过分别调整 dfs.client.hedged.read.threadpool.size 和 dfs.client.hedged.read.threshold.millis 来对 hedged read 进行测试，

所有 SQL 会跑 5 次。第一次是预热，不算。后面 4 次取平均值。

内网 starrocks ping emr 集群，平均延迟只有 0.6 ms。

实验结果单位为毫秒。

单线程查询测试

可以看出开启 hedged read 是会对性能有影响。但是开启 hedged read 后，只要参数设置的不是特别离谱，影响没那么大。

单线程查询 + 慢节点

我使得一个 HDFS 节点成为慢节点，所有请求随机延迟 0~10ms 之间。该 HDFS 节点上面没有 be 运行的。

随机延迟命令

sudo tc qdisc add dev eth0 root netem delay 5ms 5ms 25%

删除延迟

sudo tc qdisc del dev eth0 root netem

其实只要一旦存在慢节点，hedged read 一定是正优化的，只是多少的问题。

三并发，BE CPU 全部打满的情况

从结果上看起来貌似查询并发和 hedged read 之间并没有太大的影响。

总结

线程池的影响

Hedged read 是和 DFSClient 绑定在一起，说明一个 Hdfs client 会有一个 hedged read 线程池。然后那个线程池是 ThreadPoolExecutor。

Hadoop FileSystem 里面会自带一个 cache，他的 cache key 是：

  static class Key {
    final String scheme;
    final String authority;
    final UserGroupInformation ugi;
    final long unique;   // an artificial way to make a key unique

    Key(URI uri, Configuration conf) throws IOException {
      this(uri, conf, 0);
    }

    Key(URI uri, Configuration conf, long unique) throws IOException {
      scheme = uri.getScheme()==null ?
          "" : StringUtils.toLowerCase(uri.getScheme());
      authority = uri.getAuthority()==null ?
          "" : StringUtils.toLowerCase(uri.getAuthority());
      this.unique = unique;

      this.ugi = UserGroupInformation.getCurrentUser();
    }

    @Override
    public int hashCode() {
      return (scheme + authority).hashCode() + ugi.hashCode() + (int)unique;
    }

    static boolean isEqual(Object a, Object b) {
      return a == b || (a != null && a.equals(b));
    }

    @Override
    public boolean equals(Object obj) {
      if (obj == this) {
        return true;
      }
      if (obj instanceof Key) {
        Key that = (Key)obj;
        return isEqual(this.scheme, that.scheme)
               && isEqual(this.authority, that.authority)
               && isEqual(this.ugi, that.ugi)
               && (this.unique == that.unique);
      }
      return false;
    }

    @Override
    public String toString() {
      return "("+ugi.toString() + ")@" + scheme + "://" + authority;
    }
  }
}

可以看出来，这个 CacheKey 主要是根据 schema 和 authority 进行计算。这意味着在一个 BE 中，所有针对同一个 HDFS 集群的请求，都是用的同一个 DFSClient，也就是用着同一个 hedged read 线程池。

从注释上可以看出，这个线程池是动态扩缩容的，不会存在占着茅坑不拉屎。

创建 hedged read 线程池时候，这个 num 是我们配置的线程池大小，可以看出当线程 idle 超过 60s 后，就会被终止。

我自己也用 jstack 确认过，当没有 query 时，jstack 是打不出 hedged read 的线程。

推荐值

这里推荐给 dfs.client.hedged.read.threshold.millis 设置一个较大的值，比如 3000ms，这样能够有效避免极端情况。然后如果用户的 HDFS 正常的话，也不会发起 hedged read，使得数据多读。

关于 dfs.client.hedged.read.threadpool.size 的大小，实验上看不出 32/128 之间的影响，故这里推荐 128。