Kylin存储和查询的分片问题

本文来自网易云社区

作者：汪胜

相关概念介绍

为了了解Kylin存储和查询的分片问题，需要先介绍两个重要概念：segment和cuboid。相信大数据行业的相关同学都不陌生。Kylin每次提交一个新的build任务都会生成一个新的segment，而用户一般都是每天构建一次。那么，这种情况下，每天都会生成一个新的segment，用来保存昨天的数据。 Kylin的核心思想是预聚合，就是将用户预先定义的维度组合计算出来，然后保存到HBase中。这样查询的时候就可以直接查询预先计算好的结果，速度非常快。这里的维度组合就是cuboid。Kylin在构建过程中，会产生很多的cuboid数据（每一种cuboid都对应着一种维度组合），这些数据最终都会以HFile的形式存储在HBase中。Kylin对于每一个cuboid都会有一个唯一的id（一个cube的所有segment都有着相同的cuboid和cuboid id）。而这个id就是根据用户在定义cube时，维度列的排序来确定的。下面来举一个简单的例子。假设表一共有三列ABC，那么所有的cuboid组合就是：

cuboid	cuboid_id
ABC	7(111)
AB	6(110)
BC	5(101)
AC	4(100)
A	3(011)
B	2(010)
C	1(001)

其中，cube的维度列顺序为A，B，C，括号里面的是id对应的二进制，用户可以在构建cube的时候进行排序。最终数据在HBase中存储的时候，rowkey也就是按这个顺序将这些维度值组合起来（rowkey还包含其他一些成员，这里不展开）。一般推荐将用户经常使用或者基数很大的维度放在前面，这样在查询的时候有利用提高扫描效率。

存储分片问题

Kylin在build过程中，每一个cuboid的数据都会被分到若干个分片中（这里的分片就对应HBase中的region）。对于每个segment都会保存cuboidShardNums和totalShards成员。如下所示：

//key表示cuboid的id，value表示该cuboid占用的region数
private Map<Long, Short> cuboidShardNums = Maps.newHashMap();
//该segment占用的region总数
private int totalShards = 0;

请注意，一个region可能会存储多个cuboid数据，因此cuboid和region之间是多对多的关系。 Kylin可以通过下面三个配置项来控制生成build过程中生成的region相关信息：

//单个region的大小
kylin.storage.hbase.region-cut-gb
//region最小数量
kylin.storage.hbase.min-region-count
//region最大数量
kylin.storage.hbase.max-region-count

通过上面这三个配置项，我们就可以控制每个build过程中生成的region数量和大小，从而进行相应的优化。segment的分片信息也会收到这几个参数的影响。具体如下：

float cut = cubeDesc.getConfig().getKylinHBaseRegionCut();
int nRegion = Math.round((float) (totalSizeInM / (cut * 1024L)));
nRegion = Math.max(kylinConfig.getHBaseRegionCountMin(), nRegion);
nRegion = Math.min(kylinConfig.getHBaseRegionCountMax(), nRegion);
//省略余下部分代码

其中，cut就是通过kylin.storage.hbase.region-cut-gb来设置的region分割阈值，totalSizeInM是本次build过程中生成的数据大小（所有cuboid数据之和），这样就可以求出每个segment对应的totalShards大小，即nRegion。再通过如下代码便可以求出每个cuboid所占用的分片数：

int mbPerRegion = (int) (totalSizeInM / nRegion);
for (long cuboidId : allCuboids) {
  double estimatedSize = cubeSizeMap.get(cuboidId);
  double magic = 23;
  int shardNum = (int) (estimatedSize * magic / mbPerRegion + 1);
  if (shardNum < 1) {
    shardNum = 1;
  }
  //省略余下部分代码
}

首先求出每个region的实际大小mbPerRegion，然后根据每个cuboid的数据大小estimatedSize就可以求出每个cuboid所占的region数，即shardNum。这里使用了一个magic，这是为了将cuboid数据尽量分散到多个region中，这样在查询的时候就可以多个region并行扫描，提高查询效率。 搞定cuboidShardNums和totalShards之后，还需要确定每个cuboid存储数据的起始region（再通过region数shardNum便可以确定指定cuboid的所有数据分布的位置）。这里主要就是根据cuboid id和region总数来获取每个cuboid存储起始region id，具体不再展开，有兴趣的同学可以自行查看源（ShardingHash.java）。

short startShard = ShardingHash.getShard(cuboidId, nRegion);

Segment使用cuboidBaseShards成员来保存cuboid id和起始region id的映射，如下所示：

private Map<Long, Short> cuboidBaseShards = Maps.newConcurrentMap();

这样一来，就基本搞定了Kylin build过程中，segment的存储分片问题。

查询分片问题

当新的segment生成之后，我们就可以查询其中的数据了。从上面的分析中我们得知，每一个segment的构建结果其实就是多个cuboid的数据集合。那么，当我们进行查询的时候，Kylin会根据sql中的列来获取到最佳匹配的cuboid（join情况下可能会存在多个匹配的cuboid）。然后根据筛选出来的cuboid id去对应的segment中进行扫描。Kylin对于每一个待扫描的segment都会生成一个CubeSegmentScanner。在对每个segment进行扫描的时候，首先需要根据筛选到的cuboid id去获取相应的region信息（主要是起始region id和region数）。主要处理逻辑如下所示：

//传入的三个参数都可以通过cuboid id去相应的segment中获取
private List<Pair<byte[], byte[]>> getEPKeyRanges(short baseShard, short shardNum, int totalShards) {
  if (shardNum == 0) {
    return Lists.newArrayList();
  }

  if (shardNum == totalShards) {
    //该cuboid的数据分布在所有的region中
    return Lists.newArrayList(Pair.newPair(getByteArrayForShort((short) 0),
               getByteArrayForShort((short) (shardNum - 1))));
  } else if (baseShard + shardNum <= totalShards) {
    //该cuboid的数据分布在id连续的region中
    return Lists.newArrayList(Pair.newPair(getByteArrayForShort(baseShard),   
               getByteArrayForShort((short) (baseShard + shardNum - 1))));
  } else {
    //0,1,2,3,4 存储在 4,0这种情况
    return Lists.newArrayList(Pair.newPair(getByteArrayForShort(baseShard),    
               getByteArrayForShort((short) (totalShards - 1))), 
               Pair.newPair(getByteArrayForShort((short) 0), 
               getByteArrayForShort((short) (baseShard + shardNum - totalShards - 1))));
  }
}

private byte[] getByteArrayForShort(short v) {
  byte[] split = new byte[Bytes.SIZEOF_SHORT];
  BytesUtil.writeUnsigned(v, split, 0, Bytes.SIZEOF_SHORT);
  return split;
}

这样就可以获取每个segment需要扫描的region，由于Kylin目前的数据都存储在HBase当中，因此扫描的过程都在HBase中进行。对于每个region，kylin都会启动一个线程来向HBase发送扫描请求，然后将所有扫描的结果返回，聚合之后再返回上一层。为了加快扫描效率，Kylin还使用了HBase的coprocessor来对每个region的扫描结果进行预聚合。关于coprocessor的相关知识这里就不再介绍，可参考源码（CubeHBaseEndpointRPC.java和CubeVisitService.java）。
到这里，关于Kylin存储和查询的分片问题就整理的差不多了，本文省略了一些Kylin在使用HBase进行存储时的一些相关细节，后续会陆续补充上来，有感兴趣的同学可以一起交流学习。

网易云免费体验馆，0成本体验20+款云产品！ 

更多网易研发、产品、运营经验分享请访问网易云社区。