Kylin Cube构建步骤剖析&优化点_kylin.hbase.hfile.size.gb

Kylin目前作为公司离线OLAP的核心组件，在线上稳定运行已经一年之久，在这之间遇见过各种问题：比如查询慢、cube构建不出来等等；遇到诸如此类的问题时候，通常需要对cube进行调优，而对于cube的各个构建步骤的理解是作为cube调优过程中的必备技能之一，本篇文章将从cube的构建步骤出发，结合原理来谈谈在各构建步骤的cube优化该如何抉择

1.1 Create Intermediate Hive Table

• 创建Hive的临时平表

• 作为构建的第一步，Kylin从Hive中将源数据抽取出来并插入到一张临时创建的表中

• 后续的处理将以这张表为输入，待cube构建完成后，再将其删除

• 抽取时只会选择cube模型中用到的列，如果模型是按某个日期／时间列做分区的，还会将时间条件应用在创建平表的Hive命令中

• 下面是一个创建Hive平表的示例语句，可以通过kylin web ui的monitor界面进行查看：

  hive -e “USE default;
  DROP TABLE IF EXISTS kylin_intermediate_airline_cube_v3610f668a3cdb437e8373c034430f6c34;
  
  CREATE EXTERNAL TABLE IF NOT EXISTS kylin_intermediate_airline_cube_v3610f668a3cdb437e8373c034430f6c34
  (AIRLINE_FLIGHTDATE date, AIRLINE_YEAR int, AIRLINE_QUARTER int, …, AIRLINE_ARRDELAYMINUTES int)
  STORED AS SEQUENCEFILE
  LOCATION ‘hdfs:///kylin/kylin200instance/kylin-0a8d71e8-df77-495f-b501-03c06f785b6c/kylin_intermediate_airline_cube_v3610f668a3cdb437e8373c034430f6c34’;
  SET dfs.replication=2;
  SET hive.exec.compress.output=true;
  SET hive.auto.convert.join.noconditionaltask=true;
  SET hive.auto.convert.join.noconditionaltask.size=100000000;
  SET mapreduce.job.split.metainfo.maxsize=-1;
  
  INSERT OVERWRITE TABLE kylin_intermediate_airline_cube_v3610f668a3cdb437e8373c034430f6c34 SELECT
  AIRLINE.FLIGHTDATE, AIRLINE.YEAR, AIRLINE.QUARTER,…, AIRLINE.ARRDELAYMINUTES
  FROM AIRLINE.AIRLINE as AIRLINE
  WHERE (AIRLINE.FLIGHTDATE >= ‘1987-10-01’ AND AIRLINE.FLIGHTDATE < ‘2017-01-01’);”
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17

• 优化点:

  • 默认是SequenceFile的，是可以考虑改用ORC进行存储；从官网给出的数据来看是SequenceFile的效率更高
  • 可以在这步打开小文件的合并，默认是关闭的

1.2 Redistribute Intermediate Table

1.2.1 随机分布

• 重分布数据

• 在第一步完成后，Hive将抽取出的数据生成在指定的HDFS目录下

• 我们仔细观察后可以发现，有的文件比较大，有的文件比较小甚至是空的

• 文件大小的不均衡，会导致后续的MR任务也不均衡:

  • 有的Map分到的文件块较大所以耗时较长
  • 有的分到的很小所以耗时较短

• 这样就会形成“木桶效应”，为消除这种不均衡，Kylin需要对临时表的数据做一次重新分布，希望借助这次重分布使得文件块的大小适中且基本一样大，根本原理是通过DISTRIBUTE BY 来解决的

• 下面是一个重分布数据的例子:

  total input rows = 159869711
  expected input rows per mapper = 1000000
  num reducers for RedistributeFlatHiveTableStep = 160
  
  Redistribute table, cmd: 
  
  hive -e “USE default;
  SET dfs.replication=2;
  SET hive.exec.compress.output=true;
  SET hive.auto.convert.join.noconditionaltask=true;
  SET hive.auto.convert.join.noconditionaltask.size=100000000;
  SET mapreduce.job.split.metainfo.maxsize=-1;
  set mapreduce.job.reduces=160;
  set hive.merge.mapredfiles=false; 
  INSERT OVERWRITE TABLE kylin_intermediate_airline_cube_v3610f668a3cdb437e8373c034430f6c34 SELECT * FROM kylin_intermediate_airline_cube_v3610f668a3cdb437e8373c034430f6c34 DISTRIBUTE BY RAND();
  “
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16

• kylin是通过查询Hive获得临时表的行数，基于此行数计算出需要重分布的文件块数量

• 默认情况下kylin会为每100万行数据分配一个文件：

  • 在此例中，临时表有接近1.6亿行数据，所以预计需要160个文件
  • 通过在执行INSERT INTO … DISTRIBUTE BY 的重分布语句前设置“mapreduce.job.reduces=N”的方式，使用指定数量的reducer来接收数据
  • 按照每个reducer写一个文件的惯例，从而得到指定数量的文件
  • 在后续以此临时表为输入的MR处理中，Hadoop会启动与文件数相同的Map（通常100万行数据的大小不超过一个HDFS文件块）
  • 因此可以通过这一步，既合并了小文件，也消除了木桶效应

• 每个文件100万行的配置是一个经验值；实际的设置需要预估每条业务线的数据量来决定的

• 可以通过在conf/kylin.properties中设置特定参数，如:kylin.job.mapreduce.mapper.input.rows=500000

• 多数情况下，Kylin会让Hive随机地重分布数据，因此我们也可以看到重分布语句是

  DISTRIBUTE BY RAND()；因为是随机的，它会让最终的数据分布的非常均匀，且毫无规律

1.2.2 指定字段进行重分布(优化手段)

当然，我们也可以通过指定字段来进行数据的重分布，只需要在RowKey设计界面时，对需要指定的字段选择Shard By即可，值得注意的是此列需要是一个高基数列：

• 高基数可以理解为对维度count distinct之后获取的值，该值越高，则基数越高

• 按高基数维度列做重分布的好处是，具有相同此列值的行，将被写到同一个分片当中去

• 这样就非常利于后续的聚合计算：大量的聚合在本地完成(可以理解为combine)，减少了数据的交叉传递

• 在一个典型场景中，这一优化使得cube构建的时间减少了40%。

设置高基数维度列时的注意点:

• shard by列必须是一个高基数列，并且它会出现在很多cuboid中，而不是仅仅出现在少数cuboid中：
  • 这样做是为了让数据合理进行分发，并且可以在每个时间范围内的数据均匀分布，否则会造成数据倾斜，从而降低build效率
• 仅支持选择一个列作为shard by
• 使用shard by对于cube的存储和查询也有益处: 同一种类型的数据存储在一块了，根据该维度列作为筛选条件或是group by查询的时候，效率肯定比散落在各个文件当中要快的

按高基数维度进行分片会提高查询速度的原理：

• 当选取了一个维度用于分片后，如果cuboid中的某两行在该维度上的值相同，那么无论这个cuboid最终被划分成多少个分片，这两行数据必然是会被分配到同一个分片文件中去的
• kylin在读取cuboid数据的时候会先向存储引擎的若干个机器发送读取的RPC请求
• 在RPC请求的接收端，存储引擎会预先读取本机的分片数据，并先进行一定的预处理(本地预聚合)后再发送RPC回应
• 这种分片策略对查询有着极大的好处；以HBase存储引擎为例，不同的Region代表了不同的Cuboid分片
• 在读取cuboid的数据的时候，HBase会为每个Region开启一个Coprocessor实例来处理查询引擎的请求；查询引擎将查询条件和group by的条件作为请求参数的一部分发送到Coprocessor中，Coprocessor就能够在返回结果之前对当前分片的数据进行一定的预聚合(非cube构建时的预聚合，是针对查询的预聚合)

1.3 Extract Fact Table Distinct Columns

• 抽取各维度不同的值

• 在此步骤中kylin会启动一个mapreduce任务来获取各个维度在临时表中出现的不同值，用于构建维度字典

• 实际上该步骤做了更多事情:

  • 会模拟cube的整个构建过程并使用HyperLogLog计数器来估算每个cuboid的行数，从而收集cube的统计数据
  • 这些统计数据将在后续的任务中辅助kylin采取不同的策略和配置，而且还可以帮助用户后续分析等
  • 主要用于后续cube的自动优化(cube planner)

• 在这一步中如果发现map任务非常慢，通常就说明cube的设计会有问题:

  • 设计过于复杂，比如说有太多维度、太多组合等
  • 整个时候就需要我们对cube来进行优化

• 在这一步中如果发现reduce任务总是报OutOfMemory 错误:

  • 说明cube组合数过多或者默认的YARN内存分配过少
  • 通常会加大reduce端的使用内存或者按照上述所说的情况对cube进行优化

• 当然我们还可以通过修改配置来降低此步的数据采样率，从而使得整个cube的构建效率提升，但并不推荐这样做因为它会降低cube统计的准确性

1.4 Build Dimension Dictionary

• 构建维度字典
• 在上一步完成后，每个采用字典编码的维度，在此临时表中出现的不同值将被抽取在单个文件中
• 在此步中，kylin会通过读取这些值并在任务引擎节点的内存中构建字典（从下一版开始Kylin会将此任务移至MR，在2.X版本当中还不是一个MR任务）
• 通常此步骤会在数秒中完成；但是如果有超高基数（大于一百万）维度­，Kylin可能会报Too high cardinality is not suitable for dictionary的信息，建议用户对超高基数维度采用非字典编码，如fixed_length、integer等
• 对编码的设置在cube的高级设置页完成即可

1.5 Save Cuboid Statistics

• 保存cuboid统计信息
• 将采集的各个cuboid的统计信息保存至kylin的元数据存储，供后续使用

1.6 Create HTable

• 创建HTable
• 依据统计信息为当前构建的segment创建HBase table
• 1.5和1.6这2个步骤都是轻量级的任务，所以都很快可以完成

1.7 Build Base Cuboid

• 这是逐层(by-layer) vube构建的第一轮MR任务，它以临时表(即图中的“full data”)为输入，计算包含所有维度的组合(称为base cuboid)，也就是下图中的4-D cuboid:

• 此任务的map数等于步骤2的reduce数量

• 此任务的reduce数量是基于cube统计出的此任务的output大小而估算的:

  • 默认为每500MB的输出分配一个reduce
  • 如果观察到此步骤的reduce数量较小，可以考虑在conf/kylin.properties中修改默认参数以获得更多reduce同时运行， 如:kylin.job.mapreduce.default.reduce.input.mb=200
  • 这里的设置也是需要根据实际业务线的数据量来定的

1.8 Build N-Dimension Cuboid

• 构建N维的Cuboid

• 这是逐层cube构建的其它轮当中的mapreduce任务，每一轮以上一轮的输出为输入，然后减去一个维度生成新的组合，并在reduce中聚合以生成子cuboid

  • 例如: 以ABCD组合为输入，减去维度A生成BCD组合，减去维度B生成ACD组合

• 有些维度组合可以从多个父亲组合来生成，在这种情况下，Kylin会选择ID值“最小”的那个父亲组合来生成:

  • 例如，组合AB可以从ABC(id: 1110)生成，也可以从ABD(id: 1101)生成，由于1101 < 1110, 所以ABD会被用来生成AB

  • 基于此规则，如果维度D的基数很小，那么此时的聚合量将会很少，也就更加高效

  • cuboid生成树:

• 因此，当我们在设计cube的时候，在高级设置页面，请将低基数维度尽量拖动放置在高基数维度的后面；这个原则不仅会使得cube构建更快，也会使得查询时的后聚合性能更好:

  • 以ABCD为例，D的基数是最小的，因此和C相比，同时与AB组合成的维度组合，对应的id也就越小

• 通常从N-维到 (N/2)-维的计算比较慢，因为这是一个数据爆炸的过程(可以通过cuboid生成树来进行理解)：

  • N-维有1个cuboid
  • (N-1)-维有N个 cuboid
  • (N-2)-维有N*(N-1)个 cuboid
  • 以此类推…
  • 在 (N/2)-维之后，组合数会呈现出一个相反的过程，并且每个cuboid的行数逐渐减少，因此每步的build时间逐渐缩短
  • 需要注意的是，对于已经计算得出的cuboid，kylin不会重复计算

1.9 Build Cube

• 使用in-mem算法构建cube
• 这一步是采用kylin的“in-mem” 算法来构建cube:
  • 仅需一轮mapreduce即可完成所有组合的计算，但会需要使用较多的计算资源
  • kylin目录下的 conf/kylin_job_conf_inmem.xml配置文件就是为这一步任务而添加的
  • 默认会为每个map任务向YARN请求3GB内存，如果集群内存资源较充足，可以修改配置以分配更多内存，这样kylin在计算时可以更多地使用内存来缓存数据，减少磁盘的读写，从而提升构建性能
  • mapreduce.map.memory.mb
  • mapreduce.map.java.opts
• “in-mem” 算法和“by-layer”算法各有优劣；在有些情况下“in-mem”算法反而会慢；因此通常不建议我们自己去设置算法，kylin会根据cube的统计信息自动去选择更合适的算法

1.10 Convert Cuboid Data to HFile

• 生成HFile

• 此步骤启动一个mapreduce任务，将前面计算好的cuboid文件(sequence文件格式)转换成HBase的HFile

• Kylin会根据cube统计信息，计算并预先分配好region:

  • 默认会为每5GB的数据分配一个region，然后再为每个region分配合适数量的HFile
  • mapreduce的reduce数量等于HFile文件的数量
  • 通常region数越多，这一步的并发度就越高
  • 如果我们观察到这一步的耗时较长，且reduce数量较少的话，可以将下列参数的值设置的更小以获得更好性能，例如:
    • kylin.hbase.region.cut=2
    • kylin.hbase.hfile.size.gb=1
    • 参数的设置视实际业务数据量与运行情况而定

1.11 Load HFile to HBase Table

• 加载HFile到HBase
• 在这步中kylin使用HBase API将生成好的HFile通过BulkLoad的方式加载到HBase的region server中
• 通常这一步也是很快就能完成的

1.12 Update Cube Info

• 在cube数据加载到HBase后，kylin会将此次构建的Cube segment的状态，从”NEW”更新成”READY”，标记此segment可以供查询引擎使用
• 思考: 重刷线上的cube，是否会影响到线上的查询
• 答案: 会影响到，但是转瞬即逝，前面耗时的步骤并不会影响查询，只有在这步中才会影响到

1.13 Clean up

• 这是收尾的步骤，会删除第一步生成的临时表
• 到这里它已经不会影响其它任何任务，因为segment已经变成了READY状态
• 注意点: cube如果构建到一半失败，这步是不会被执行到的，那些中间文件是不会被清除的，因此需要使用过StorageCleanupJob来进行清理，官方提供了工具可以供我们人为进行使用，每周定时调度一次即可