为了了解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进行存储时的一些相关细节,后续会陆续补充上来,有感兴趣的同学可以一起交流学习。