Kylin Cube构建过程优化

2/13/2017来源:SQL技巧人气:2731

原文地址:https://kylin.apache.org/docs16/howto/howto_optimize_build.html

Kylin将一个cube的build过程分解为若干个子步骤,然后串行执行这些子步骤。这些步骤包括Hive操作,MR任务和其他类型的工作。如果每天都有许多cube进行build操作,那么肯定会办法加速这一过程。这里有一些建议可以参考,我们就按照build的顺序依次介绍。

Create Intermediate Flat Hive Table

该步骤会从源Hive表中抽取数据(将所有的相关表join之后的数据),并且插入到一个临时的扁平表中。如果cube是带有分区列的,Kylin将会增加一个时间条件,这样就会保证只有符合条件的数据才会被抓取。可以在日志中查看与该步骤相关的Hive命令。如下所示:

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'); "

当执行Hive命令的时候,Kylin会使用conf/kylin_hive_conf.xml中的配置项,例如,使用更少的副本数和允许Hive的mapper连接操作。如果有需要的话,也可以添加其他的配置项。

如果Cube的分区列(这里是“FLIGHTDATE ”)与Hive表的分区列是同一个列,那么对于该列进行过滤将会使Hive非常快速地剔除不符合条件的分区。因此强烈建议使用Hive的分区列(如果该分区列是日期)作为Cube的分区列。这对于数据非常多的表来说几乎是必须的,否则在执行这一步骤的时候,Hive每次都需要扫描所有的文件,会耗费很多时间。

如果你的Hive允许文件合并,你可以在“conf/kylin_hive_conf.xml”中通过配置项来禁用这一功能。因为Kylin有它自己的文件合并方法(我们将在下面介绍):

<property> <name>hive.merge.mapfiles</name> <value>false</value> <description>Disable Hive's auto merge</description> </property>

Redistribute intermediate table

经过上一个步骤之后,Hive会在HDFS的目录中生成一些数据文件,但是一些文件可能会很大,而另外一些文件可能会很小甚至是空的。文件大小分布的不均衡也会导致后续的MR任务执行的不平衡:一些mapper任务会执行的很快,而其他的mapper可能会执行的很慢。为了使这些数据分布的更均匀一些,Kylin增加了该步骤用来重新分配各个数据文件中的数据。下面是一个简单的输出:

total input rows = 159869711 expected input rows per mapper = 1000000 num reducers for RedistributeFlatHiveTableStep = 160

重分配数据之后执行的Hive命令如下所示:

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(); "

首先,Kylin会获取临时表的行数;然后,基于这个行数就可以获取需要进行数据重分配的文件的数量。Kylin默认一个文件包含1百万行的数据。在这个例子中,一共有1.6亿行数据和160个reducer,并且每个reducer写一个文件。在后续的MR步骤中,Hadoop将会启动相同数量的mapper来对这些文件进行处理(通常1百万行数据的大小会比一个HDFS块要小)。如果日常的数据规模不是很大或者Hadoop集群有足够的资源,可以通过在“conf/kylin.properties”中将配置项“kylin.job.mapreduce.mapper.input.rows”设置为更小的值来获取更高的并发度,如下所示:

kylin.job.mapreduce.mapper.input.rows=500000

其次,Kylin通过运行一个HiveQL “INSERT OVERWIRTE TABLE … DISTRIBUTE BY …” 来为指定数目的reducer重新分配数据行。

在大多数情况下,Kylin会要求Hive随机地为这些reducer重新分配数据行,以保证这些文件在大小上相近。此时,重分配语句就是 “DISTRIBUTE BY RAND()”

如果Cube中确定了一个“shard by”维度列(在Cube的“Advanced setting”界面进行设置),该维度列是一个基数很大的维度列(例如“USER_ID”),那么Kylin会要求Hive通过该维度列的值来重新分配数据。对于该列拥有相同值的数据行将会被分配在同一个文件中。这比随机分配要好,因为数据不仅被重新分配了,而且在没有消耗额外成本的情况下就对数据进行了重新分类,这对于后续的Cube构建过程是有帮助的。在特定的情况下,这种优化可以减少40%的构建时间。这种情况下的重分配语句就是 “DISTRIBUTE BY USER_ID”

请注意: 1.设置为“shard by”的维度列应该是一个基数很大的维度列,并且它会出现在很多cuboid中(不会是仅仅出现在很少的cuboid中)。使用它进行重分配可能会在每一个时间间隔上获得等分布;反之则会造成数据倾斜,而这则会减少构建速度。典型的适合情景就是:“USER_ID”,“SELLER_ID”,“PRODUCT”,“CELL_NUMBER”等等,一般基数应该大于1000(应该远远大于reducer的数目)。 2.使用“shard by”在Cube存储中还有其他的优势,但不在本文的讨论范围中。

Extract Fact Table Distinct Columns

在此步骤中,Kylin通过运行一个MR任务来获取维度列的distinct值,用于进行字典编码。

实际上,该步骤还做了其他的工作:通过使用HyperLogLog计数器预估每个cuboid的行数,依次来收集cube的统计信息。如果你发现mapper任务执行非常慢,通过就意味着cube设计的太复杂,可以参考:Cube设计优化,对cube进行优化,使cube更加精简。如果reducer发生了OOM错误,通常意味着cuboid的维度组合数太多或者默认的yarn内存分配不能满足需求。如果此步骤不能在合理的时间内完成,请重新对cube进行设计,因为真正的build过程会花费更长的时间。

你可以减少采样比例(通过在kylin.properties中设置kylin.job.cubing.inmem.sampling.percent配置项),来加速该步骤的执行,但是这可能不会有太大的效果,而且还会影响cube统计信息的准确性,因此一般不推荐这么做。

Build Dimension Dictionary

在上一步中已经获得了所有维度列的distinct值,接着Kylin将会在内存中构建字典(在下个版本中将会将此操作移到MR任务中)。通常这一步会很快,但是如果distinct值的集合很大,Kylin可能会报错,例如,“Too high cardinality is not suitable for dictionary”。对于UHC(超大容量)列,请使用其他编码方式,例如“fixed_length”,“integer”等。

Save Cuboid Statistics and Create HTable

这两个步骤是轻量级的,并且很快。

Build Base Cuboid

这一步是通过临时表构建基本的cuboid,这是逐层算法的第一轮MR任务。Mapper的数量等于步骤二中reducer的数量;而Reducer(这里指的是本步骤中启动的reducer)的数量是通过cube的统计信息预估出来的:默认每500M使用一个reducer。如果你发现reducer的数目很少,可以通过在kylin.properties中对配置项“kylin.job.mapreduce.default.reduce.input.mb”设置更小的值,来获取更多的集群资源,如下所示: kylin.job.mapreduce.default.reduce.input.mb=200

Build N-Dimension Cuboid

这些步骤是逐层算法的处理过程,每一步都使用前一步的输出作为输入,然后去除某个维度进行聚合,生成一个子cuboid。例如,对于cuboid ABCD,去除维度A可以获得cuboid BCD,去除维度B可以获得cuboid ACD等。

有些cuboid可以通过一个以上的父cuboid聚合而成,在这种情况下,Kylin将会选择最小的父cuboid。例如,AB能够通过ABC(id:1110)和ABD(id:1101)聚合生成,因此ABD会被作为父cuboid使用,因为它的id比ABC要小。基于以上处理,如果D的基数很小,那么此次聚合操作就会花费很小的代价。因此,当设计cube的rowkey顺序的时候,请记住,将低基数的维度列放在尾部。这不仅对cube的构建过程有好处,而且对cube查询也有好处,因为后聚合(应该是指在HBase查找对应cuboid的过程)也遵循这个规则。

通常从N-D到(N/2)-D的构建过程很慢,因为这是一个cuboid爆炸增长的过程:N-D有1个cuboid,(N-1)-D有N个cuboid,(N-1)-D有N*(N-1)个cuboid等等。在(N/2)-D步骤之后,构建过程会越来越快。

Build Cube

这一步骤使用了一种新的算法来构建cube:逐块算法(也称作“in-mem”算法)。该算法只使用一轮MR任务来构建所有的cuboid,但它比逐层算法需要更多占用更多的内存。该步骤在执行的时候会使用“conf/kylin_job_conf_inmem.xml”中的相关配置项。默认情况下,每个mapper需要3G的内存。如果集群有足够大的内存,可以在“conf/kylin_job_conf_inmem.xml”中通过修改配置来获取更大的内存,这样就可以处理更多的数据,并且性能也会更好。修改配置如下所示:

<property> <name>mapreduce.map.memory.mb</name> <value>6144</value> <description></description> </property> <property> <name>mapreduce.map.java.opts</name> <value>-Xmx5632m</value> <description></description> </property>

请注意,Kylin会根据数据分布(通过cube统计信息获取)自动地选择合适的cube构建算法。一般不需要显示地选择使用哪个算法。

Convert Cuboid Data to HFile

这一步会启动一个MR任务用来将cuboid文件(顺序文件格式)转换为Hbase的HFile文件。Kylin通过cube的统计信息来计算HBase的region个数,默认每个region大小是5G。Region数越多,就会使用更多的reducer。如果发现reducer的数目很少,并且性能很差,就可以在“conf/kylin.properties”中增加如下配置项:

kylin.hbase.region.cut=2 kylin.hbase.hfile.size.gb=1

如果不能确定一个HBase的region该设置为多大,请联系HBase管理员。

Load HFile to HBase Table

这一步使用了HBase API将HFile导入到HBase的region中,这一步很简单,也很快。

Update Cube Info

将数据导入Hbase中之后,Kylin会将新生成的segment在元数据中的状态修改为ready。这一步也非常快。

Cleanup

这一步主要就是从Hive中删除临时表。由于在上一步中,已经将segment的状态修改为ready,所以这一步的操作不会对segment产生任何影响。即使这一步执行发生了错误,也不需要担心,因为所有的垃圾都会在Kylin执行StorageCleanupJob的时候进行回收。