关注交流微信公众号：小满锅

场景

前几天遇到一个任务，从前也没太注意过这个任务，但是经常破9点了，执行时长正常也就2个小时。
看逻辑并不复杂，基本是几段SQL的JOIN操作，其中一个最耗时间的就是要根据底表数据Lateral view explode(array(字段, ‘all’))，一共lateral了4个字段，相当于数据量要扩大16倍。并且可怕的场景，lateral view之后还对11个字段进行了去重。

select
    a22 as a,
    b22 as b,
    c22 as c,
    d22 as d,
    e,
    f,
    ,count(distinct g) as g    
    ,count(distinct h) as h  
    ,count(distinct i) as i    
    ,count(distinct j) as j  
    ,count(distinct k) as k
    ,count(distinct l) as l
    ,count(distinct m) as m
    ,count(distinct n) as n
    ,count(distinct o) as o
    ,count(distinct p) as p
    ,count(distinct q) as q
    ,count(distinct r) as r
    ,count(distinct s) as r
from 
(
    select 
        a
    ,b
    ,c
    ,d
    ,e
    ,f,g,h,i,j,k,l,...,r,s
    from table
    where dt='${etl_date}') t
    lateral view explode(array(a, 'all')) a as a22
    lateral view explode(array(b, 'all')) b as b22
    lateral view explode(array(c, 'all')) c as c22
    lateral view explode(array(d, 'all')) d as d22
    group by 1,2,3,4,5,6,7,8,9,10,11,12,13
) t
group by a,b,c,d,e,f

定位问题

查看了执行计划，发现这个任务从Input：4亿多条数据 竟然expand到了700多亿，太夸张了，还以为Spark3.1.1能对这种多维分析场景有很好的优化呢。

但是我又百思不得其解了，从这个数据量来看，应该是一共扩大了176倍，其中Lateral View扩大了16倍我可以理解，剩下的11倍哪里来的？
突然觉得11这个数字很熟悉，就是那11个count(distinct)字段。仔细查看了执行计划，发现除了lateral view 有expand之外，Spark在对Count(distinct)处理时，根据11个字段的去重字段，对于每条记录会都会expand出11条记录，假设为A1-A11，那么每一条记录都对应唯一一个A字段有值，其他10个A字段为NULL，而count(distinct A)只需要计算count(1)即可，看似应该是为了结束数据倾斜的一个优化。这样就解释的通了。
相当于从Read->filter->expand->shuffle 原来的一条记录，变成了176条，也就是扩大了176倍。

解决问题

这种现象第一时间想到的是从两个方面入手，尽量减少这个任务在这个Stage消耗的时间。

• 第一种就是增加partition，这样减少每个partition的运行时间。
• 第二种就是减少expand的数据量，从根本上取解决问题。

第一种方案

设置参数
conf.spark.sql.shuffle.partitions = 4096
conf.spark.sql.files.maxPartitionBytes = 8m
conf.spark.shuffle.minNumPartitionsToHighlyCompress = 5000
conf.spark.sql.finalStage.adaptive.skewJoin.skewedPartitionFactor = 3

可以看到partition确实增加了，但是留有的风险比较多，这种就是要通过加资源保证任务快速运行，实际上这个任务不是那种P0任务，这个21min至少得保持1k多的task在并行跑才行，而实际任务在调度的时候，不可能抢到这么多资源，并且不能影响线上重要任务运行。因此这个方案暂时不去做深入研究了。

第二种方案

第二种方案是通过减少expand的量，这样下游处理的数据就会更少一些了。

尝试一

我原本想通过减少expand的量去解决这种问题，比如在子查询中聚合到最细的粒度，然后再去lateral view，但是在本地自己查询发现，即便是去重到最细粒度，仍然有4亿多条，经过那176倍之后，仍然会有700多亿条数据。这个想法失败。

尝试二

因为16个聚合多维分析这个必须存在，所以说扩大16倍是避免不了的(或者有啥其它办法也说不定呢)。

想办法从11个去重字段去入手，可以尝试分别分为两个stage各计算一半的字段，然后结果join起来，但是看起来会需要资源的支撑。

也可以看可不可以关掉count(distinct)的这个数据倾斜优化，找了半天没找到。

尝试三

改用bitmap对字段进行去重 大致就是自己写了一个UDAF，对于整个stage的每个阶段，实现UDAF抽象类的一些方法，比如处理原始数据，输出给下游采用byte[]形式，shuffle时merge，最终聚合输出。这里中间存储的数据结构为RoaringBitMap，听说这个存储和性能比较好，并且shuffle合并聚合时，可以直接利用它的OR操作进行位去重即可，快速且方便。

*   BitmapDistinctUDAFBuffer 约定map和reduce的每个处理过程
 *   BitmapAggrBuffer 作为中间结果处理器。
 *
 *   COMPLETE阶段：只要map阶段才调用
 *   PARTIAL1阶段：轻度汇总传给下游
 *      init方法         : 约定原始输入数据类型为Text 下游输出为byte[]
 *      iterate方法      : 将原始输入数据都拿过来 将其写入聚合去重中间结果处理器BitmapAggrBuffer
 *      terminatePartial: 将处理过后的BitmapAggrBuffer里面的RoaringBitmap处理成byte[]输出给下游
 *   PARTIAL2阶段：shuffle之后聚合
 *      init方法         : 约定下游输出为byte[]
 *      merge           : 将iterate处理后的BitmapAggrBuffer稍微按照Group Key轻度聚合，即对不同的BitmapAggrBuffer里面的RoaringBitMap进行OR操作
 *      terminatePartial: 将处理过后的BitmapAggrBuffer里面的RoaringBitmap处理成byte[]输出给下游
 *   FINAL阶段   ：最终结果聚合
 *      init方法         : 约定下游输出为byte[]
 *      merge           : 将iterate处理后的BitmapAggrBuffer稍微按照Group Key轻度聚合，即对不同的BitmapAggrBuffer里面的RoaringBitMap进行OR操作
 *      terminate: 将处理过后的BitmapAggrBuffer里面的RoaringBitmap处理成byte[]输出
 */

资源非常充足的情况下，跑了很多次，基本30-40min跑完，申请不到资源仍然有50多min，这几次问题的共同点就是卡在了最后一个shuffle的时候，多维分析存在的问题就是，一旦分析维度存在较大差异，那么最细粒度和最粗粒度的数据一定会存在倾斜，可想而知，聚合粒度最粗的数据，要收集上游expand之后的所有数据，必定会导致这个分区长尾，执行时间也自然最长，并且这里打散还不能这样，多维分析时，得对最细的数据打散，一旦聚合到最细粒度之上了，最细粒度之下的数据无法计算，所以聚合到userid粒度时，再去打散，和上面Spark对其的优化一样，打散了多少呗，本身四亿多条数据，扩大16倍，再打散又扩大几倍，效果也不明显。

尝试四

第四种方式来源于第三种开发UDAF的时候看到RoaringBitMap可以进行一个OR操作，那么其实我读取数据的时候，上面一种方法是只能聚合到用户粒度，那聚合到用户粒度是为了多维分析时能够进行去重，那如果是基于这样的需求的话，我仍然可以聚合到多维分析的最细粒度，将用户信息存储到这个bitmap里面，就像下面这个图，很多用户id，标志0或者1，存储到RoaringBitMap数据结构中，那么下游多维分析时，只需要对轻度聚合之后的数据进行一个OR操作就可以了。

基于上面的方案，我又写了一个BitMapMerge的UDAF，主要是对BitMapDistinct出来的中间聚合结果(最细粒度的bitmap)进行合并操作(OR)。

心急如焚的跑了下，效果非常明显，在多维分析Expand之前，轻度聚合4亿多条到多维分析的最细粒度之后，发现只有3000多条数据了，效果非常明显，心想着如果执行lateral view expand之后应该也就是48000左右的数据，然后再聚合到多维分析的各个粒度，这样即便是数据倾斜了，应该也没事，处理的数据量比较小。这样从700亿数据到48000w左右的expand量，数据量上优化了145w倍

。。。完了，内存爆了，失败了，加到Excutor 12G都不行，而且发生在了Input Read的阶段，这里困扰了我，看了很久，发现一个问题就是，轻度聚合的数据只有3000左右了但是足足有11G？我直接计算出每条记录每个bitmap的数值，发现有的去重结果中，有个别千万级别有亿级别的去重数，所以想会不会就是这些个别的大数导致那一部分task失败了呢(因为我即便加了partition也没用，所以我猜测光一条数据就很大)。这个级别的去重数并不可怕，算一算即便是3亿的去重结果，按照bitmap来算的话，也就是300000000/8/1024/1024=35M左右的数据大小，但是越想越不对，就是这里有11个字段，每个都这么大，在expand，351116=6g，一条记录就要expand这么大的数据，那一个task读几条再加expand还不得起飞，这下我知道了，我遇到的不是数据量的倾斜，而是数据大小的倾斜，极个别数据太大了，导致input阶段在expand的时候超内存了。

为了验证自己的想法没错，我将11个字段减少几个比较大的bitmap字段，结果5min就跑完了，并且内存保持的相对不错这也太好了吧。于是就打算将多个结果拆分出来计算再JOIN了。

其它方案

其实为了解决上面expand的数据大小太大的问题，我们终究就是去一个数据量的倾斜和数据大小倾斜之间的一个平衡。加入我不进行轻度聚合，那么就是数据量的倾斜，假如进行轻度聚合，那么就是数据大小的倾斜。那其实就是为了这个平衡的话，可以在轻度聚合的时候，采用一个比较传统的方式，打散聚合，这样就可以平衡掉某条记录因为读取记录大小太大，导致expand出去的数据太大，假设打散100倍，那么最终expand出去的数据也就480w条，优化效果仍然比以前的700亿的数据量要好很多。

但是最终的数据还是一个task在处理，这样的话最粗粒度的key，将会聚集最大bitmap，处理时间会更久，导致长尾。

总结
对于多维分析切去重计数场景，需要尽可能减少expand的量，同时使用bitmap也有坑，一般来说，像这种多维分析场景中，粒度越粗的key总会是聚集用户数最多的，造成数据大小的倾斜，那么处理的时间也是最长的，直观上从平时打散的角度并不会有太多效果，最终处理的结果还是一样。

比较好的方式应该是拆分字段计算，最终再JOIN。最终加起来在资源不足的情况下差不多20min左右，比起资源不足的源代码至少得运行2小时-2个半小时，效果还是非常明显的，并且expand的数据量也从700亿降到了48000数据量。不过单条数据的数据大小确实变大了。

反思

这里利用BitMap分析多维场景，存在一个问题，就是说更粗粒度的计算过程中，其实已经经历了比它更细粒度的bitmap合并。这就是为什么要在内部先聚合到多维分析的最细粒度，来减少重复合并的这个过程。
那其实对于多维分析，其实每个维度本身也是嵌套的父子关系

比如分析维度
os, appver, source
那假如要分析os，appver的话其实就可以基于上面这个维度去进一步merge
加入要分析os维度，就可以基于os, appver再进一步分析。

那我优化的方式使用分JOIN计算Bitmap字段其实对于每个字段而言，还是存在数据大小的倾斜，就是说即便是一个字段，那么它的最粗粒度的分析一定是倾斜的，因为Group sets或者lateral view是将原始数据直接expand16倍，再进行聚合，那么最粗粒度一定是数据量最多(这里虽然不足48000)，自然处理数据大小也最大，可能多大几个G，数据大小太大的原因就是因为BitMap存储还是占用太多空间了，其中一个更好的优化，要是可以基于最细粒度自由上卷下钻就好了，就好比一些外部存储一样，存储最细粒度，自动帮你上卷到最粗粒度，一层层叠加，不重复计算。
那Spark暂时没想到避免重复计算的方式，后面就从改善Bitmap存储大小入手，考虑用HLL优化试试，期待能够优化到10min以内。

1 介绍一下你们的项目

教育数仓解决的问题：
首先，受互联网+概念影响，越来越多的教育平台机构涌现，在线教育发展火热。但是由于信息的共享利用不充分，导致企业多年积累了大量数据，而因为信息孤岛的问题，一直没有对这些数据进一步挖掘分析，因此也不能给企业的管理决策层提供有效的数据支撑。
有鉴于此，我们做的这个教育大数据分析平台项目，将大数据技术应用于教育行业，用擅长分析的OLAP系统为企业经营提供数据支撑。
具体的实现思路是，先建立企业的数据仓库，把分散的业务数据预处理，其次根据业务需求从海量的用户行为数据挖掘分析，定制出多维的数据集合，形成数据集市，供各个场景主题使用，最后用BI工具，进行前端展示。
所以，我们的技术解决了企业的三大痛点。一是数据量太大问题，传统数据库无法满足；二是系统多，数据分散问题，无法解决数据孤岛问题；三是，统计工作量太大，分析难度高问题，无法及时为企业提供数据参考。

2 数仓架构是什么 用到了哪些技术？

，底层数据存储在Hive，数据计算使用Spark

3 什么是全量数据？

4 什么是增量数据？

5 增量数据如何同步？

6 缓慢渐变维问题如何解决的？

7 你们数仓分了几层？

<

1、BI

1.1 BI 技术

2、OLAP基本操作和类型

2.1 OLAP基本操作

2.2 OLAP分类

• MOLAP，基于多维数组的存储模型，也是OLAP最初的形态，特点是对数据进行预计算，以空间换效率，明细和聚合数据都保存在cube中。但生成cube需要大量时间和空间。
• ROLAP，完全基于关系模型进行存储数据，不需要预计算，按需即时查询。明细和汇总数据都保存在关系型数据库事实表中。
• HOLAP，混合模型，细节数据以ROLAP存放，聚合数据以MOLAP存放。这种方式相对灵活，且更加高效。可按企业业务场景和数据粒度进行取舍，没有最好，只有最适合。

3、OLAP数据库选型

3.1 Presto

3.1.1 概念

3.1.2 presto架构（master+slaver模式）

3.1.3 Presto应用场景

3.2 Druid 

3.2.1 概念

3.2.2 Druid架构

3.2.3 基本特点

    Apache Druid 具有以下特点：

• 亚秒级 OLAP 查询，包括多维过滤、Ad-hoc 的属性分组、快速聚合数据等等。
• 实时的数据消费，真正做到数据摄入实时、查询结果实时。
• 高效的多租户能力，最高可以做到几千用户同时在线查询。
• 扩展性强，支持 PB 级数据、千亿级事件快速处理，支持每秒数千查询并发。
• 极高的高可用保障，支持滚动升级。

3.2.4 应用场景

• 实时指标监控
• 推荐模型
• 广告平台
• 搜索模型

3.2.5 Druid案例

    知乎：技术选型上，知乎根据不同业务场景选择了HBase 和 Redis 作为实时指标的存储引擎，在OLAP选型上，知乎选择了Druid。

    OPPO：而OPPO根据自身不同的业务场景，报表层选择了Druid，标签选择了ES，接口层选择了Hbase。

3.3 Kylin

3.3.1 概述

3.3.2 kylin特性

• 可扩展超快olap引擎，Hadoop/Spark上百亿数据规模
• 提供 Hadoop ANSI SQL 接口
• 交互式查询能力，用户可以与Hadoop数据进行亚秒级交互
• 百亿以上数据集构建多维立方体（MOLAP CUBE）
• 与BI工具无缝整合，如Tableau，PowerBI/Excel，MSTR，QlikSense，Hue和SuperSet

3.3.3 kylin生态圈

4、Clickhouse 

4.1 概述

4.2 场景特征

• 大多数是读请求
• 数据总是以相当大的批(> 1000 rows)进行写入
• 不修改已添加的数据
• 每次查询都从数据库中读取大量的行，但是同时又仅需要少量的列
• 宽表，即每个表包含着大量的列
• 较少的查询(通常每台服务器每秒数百个查询或更少)
• 对于简单查询，允许延迟大约50毫秒
• 列中的数据相对较小：数字和短字符串(例如，每个URL 60个字节)
• 处理单个查询时需要高吞吐量（每个服务器每秒高达数十亿行）
• 事务不是必须的
• 对数据一致性要求低
• 每一个查询除了一个大表外都很小
• 查询结果明显小于源数据，换句话说，数据被过滤或聚合后能够被盛放在单台服务器的内存中

4.3 clickhouse自身限制

• 不支持真正的删除/更新支持 不支持事务
• 不支持二级索引
• 有限的SQL支持，join实现与众不同
• 不支持窗口功能
• 元数据管理需要人工干预维护