08离线数仓

离线数仓

离线数仓分层的原因和背景

参考大厂的做法，主要是ali的云上数仓解决方案
数仓分层的好处
- 提高数据计算中间计算结果的复用性
- 方便的数据的管理
为业务数据做服务，更好的开展业务

数仓的数据来源和数据产出

前端埋点的用户行为数据，业务的数据库
爬虫，提供的api接口
指标报表，用户画像，推荐系统，风控系统，算法预测

常见数仓组件的版本

开源的组件版本

hadoop 3.0.0 2018年，后面升级到3.1.3
flume 1.9.0 2019年
kafak 2.4.0 2019年
hive 3.1.2 2019年
hbase 2.0.0 2018年
dolphsheduler 1.2.0 2020年
flink
spark 2.4.0 2018年商业版CDH
CDH 6.3.2

集群云服务器

单台配置，128G，32C，磁盘20T
选择的aliyun的云服务EMR，单台年费大概在5W左右 [[集群规模及部门介绍#^a46fb1|hadoop集群规模]]

数仓建模

Dim建模

维度建模对比关系型数据库建模ER，ER建模更多是降低数据的冗余，提供数据的一致性；而维度建模，更多的是应对大数据场景下的查询效率的提升；常见的维度模型有星型模型，雪花模型，星座模型。
- 星型模型：事实表直接和一级维度关联，一级维度表间没有关联；优点是减少了事实表和维度表的join，大大提高了查询的效率、缺点是维度与维度间没有关系描述，这种关系都体现在维度数据中，如果没有较少的管理模型，年深日久可能维度数据将会丢失。
- 雪花模型：多个维度表直接相会关联，并于事实表关联。优点是减少了维度数据的冗余
- 星座模型：是星型模型的扩展，星座模型间有多张事实表，不同的事实表间通过维度关联，通常适用于复杂的场景

dim建模步骤

plaintext
- 确定主维表和相关维表，主维表一般与事实表直接相关，其他的围绕业务过程的就可选为相关维表；维度字段一般需要多个维表相会关联，或者维度属性拆解，维度属性合并等
- 维度需要尽可能丰富
- 维度属性应该包含属性的描述

维度建模注意的问题

多值维度

比如sku的销售属性，一个sku就可能有多行的维度数据，比如颜色，和版本；此时在维度数据加载时，可以可以多行变一行的hive函数，collect_set()或者collect_list
多值属性

对于多值维度，在数据加载时使用hive的struct，或者map
维度退化
当事实表关联的维度较，比如3-4个维度，则可以选择将维度数据冗余到事实表中
缓慢变化维的处理
当面对大数据量的维度表，一般采用增量同步每日的数据，若维度数据变化较慢，但历史数据具有保留的价值，可以使用维度拉链的方式，为每一条维度数据加入生命周期（start_time,end_time），比如用户表

sql
  create temporary table dim_user_zip  
(  
    id string comment '用户id',  
    login_name string comment '用户名称',  
    nick_name string comment '用户昵称',  
    name string comment '用户姓名',  
    phone_num string comment '手机号码',  
    email string comment '邮箱',  
    user_level string comment '用户等级',  
    birthday string comment '生日',  
    gender string comment '性别',  
    create_time string comment '创建时间',  
    operate_time string comment '操作时间',  
    start_date string comment '开始日期', -- 
    end_date string comment '结束日期', -- 生命周期结束
    dt string  
)

项目中用到的维度拉链方式：

2.2 dim_user_info 实例

sql
--首日数据加载:
insert overwrite table dim_user_zip partition (dt='9999-12-31')
select
    data.id,
    data.login_name,
    data.nick_name,
    data.name,
    data.phone_num,
    data.email,
    data.user_level,
    data.birthday,
    data.gender,
    data.create_time,
    data.operate_time,
    date_format(nvl(data.operate_time,data.create_time),'yyyy-MM-dd'),
    '9999-12-31'
from ods_user_info_inc where dt='2020-06-14' and type='bootstrap-insert';
set hive.exec.dynamic.partition.mode=nonstrict;
with old as (
    select id,
           login_name,
           nick_name,
           name,
           phone_num,
           email,
           user_level,
           birthday,
           gender,
           create_time,
           operate_time,
           start_date,
           end_date
    from dim_user_zip where dt='9999-12-31'
),new as (
    --一个用户一天可能更新多次,此时只需取最新状态数据即可
    select id,
           login_name,
           nick_name,
           name,
           phone_num,
           email,
           user_level,
           birthday,
           gender,
           create_time,
           operate_time,
            '2020-06-15' start_date,
            '9999-12-31' end_date
    from (
             select data.id,
                    data.login_name,
                    data.nick_name,
                    data.name,
                    data.phone_num,
                    data.email,
                    data.user_level,
                    data.birthday,
                    data.gender,
                    data.create_time,
                    data.operate_time,
                    row_number() over (partition by data.id order by ts desc) rn
             from ods_user_info_inc
             where dt = '2020-06-15'
         ) t1 where rn=1
), full_user as (
    select old.id old_id,
           old.login_name old_login_name,
           old.nick_name old_nick_name,
           old.name old_name,
           old.phone_num old_phone_num,
           old.email old_email,
           old.user_level old_user_level,
           old.birthday old_birthday,
           old.gender old_gender,
           old.create_time old_create_time,
           old.operate_time old_operate_time,
           old.start_date old_start_date,
           old.end_date old_end_date,
           new.id new_id,
           new.login_name new_login_name,
           new.nick_name new_nick_name,
           new.name new_name,
           new.phone_num new_phone_num,
           new.email new_email,
           new.user_level new_user_level,
           new.birthday new_birthday,
           new.gender new_gender,
           new.create_time new_create_time,
           new.operate_time new_operate_time,
           new.start_date new_start_date,
           new.end_date new_end_date
    from old full join new
    on old.id = new.id
)
insert overwrite table dim_user_zip partition (dt)
--查询最新数据
select
    nvl(new_id,old_id) id,
    if( new_id is not null, new_login_name,old_login_name ),
    if( new_id is not null, new_nick_name,old_nick_name ),
    if( new_id is not null, new_name,old_name ),
    if( new_id is not null, new_phone_num,old_phone_num ),
    if( new_id is not null, new_email,old_email ),
    if( new_id is not null, new_user_level,old_user_level ),
    if( new_id is not null, new_birthday,old_birthday ),
    if( new_id is not null, new_gender,old_gender ),
    if( new_id is not null, new_create_time,old_create_time ),
    if( new_id is not null, new_operate_time,old_operate_time ),
    if( new_id is not null, new_start_date,old_start_date ),
    if( new_id is not null, new_end_date,old_end_date ),
    if( new_id is not null, new_end_date,old_end_date )
from full_user
union all
select
           old_id,
           old_login_name,
           old_nick_name,
           old_name,
           old_phone_num,
           old_email,
           old_user_level,
           old_birthday,
           old_gender,
           old_create_time,
           old_operate_time,
           old_start_date,
           cast(date_sub('2020-06-15',1) as string) ,
           cast(date_sub('2020-06-15',1) as string)
from full_user where new_id is not null and old_id is not null;
with old as (
    select id,
           login_name,
           nick_name,
           name,
           phone_num,
           email,
           user_level,
           birthday,
           gender,
           create_time,
           operate_time,
           start_date,
           end_date
    from dim_user_zip where dt='9999-12-31'
),new as (
    --一个用户一天可能更新多次,此时只需取最新状态数据即可
    select id,
           login_name,
           nick_name,
           name,
           phone_num,
           email,
           user_level,
           birthday,
           gender,
           create_time,
           operate_time,
            '2020-06-15' start_date,
            '9999-12-31' end_date
    from (
             select data.id,
                    data.login_name,
                    data.nick_name,
                    data.name,
                    data.phone_num,
                    data.email,
                    data.user_level,
                    data.birthday,
                    data.gender,
                    data.create_time,
                    data.operate_time,
                    row_number() over (partition by data.id order by ts desc) rn
             from ods_user_info_inc
             where dt = '2020-06-15'
         ) t1 where rn=1
),full_user as (
    select id,
           login_name,
           nick_name,
           name,
           phone_num,
           email,
           user_level,
           birthday,
           gender,
           create_time,
           operate_time,
           start_date,
           end_date,
           row_number() over (partition by id order by start_date desc) rn
    from (
             select id,
                    login_name,
                    nick_name,
                    name,
                    phone_num,
                    email,
                    user_level,
                    birthday,
                    gender,
                    create_time,
                    operate_time,
                    start_date,
                    end_date
             from old
             union all
             select id,
                    login_name,
                    nick_name,
                    name,
                    phone_num,
                    email,
                    user_level,
                    birthday,
                    gender,
                    create_time,
                    operate_time,
                    start_date,
                    end_date
             from new
         ) t1
)
insert overwrite table dim_user_zip partition (dt)
--查询最新数据
select id,
       login_name,
       nick_name,
       name,
       phone_num,
       email,
       user_level,
       birthday,
       gender,
       create_time,
       operate_time,
       start_date,
       end_date,
       end_date dt
from full_user
where rn = 1
union all
--失效数据
select id,
       login_name,
       nick_name,
       name,
       phone_num,
       email,
       user_level,
       birthday,
       gender,
       create_time,
       operate_time,
       start_date,
       cast(date_sub('2020-06-15',1) as string),
       cast(date_sub('2020-06-15',1) as string)
from full_user
where rn != 1

拉链表的退链处理数据错误时间 e_t

过滤删除出错的数据 start_time >= e_t
将误将end_time 为9999-12-31 的数据过期时间修改为当日，即insert overwrite table dim_user_info partition (dt) dt select ... ,当日 dt where end_time >= e_t比如 1月6号数据导入出现错误，导致本应该在1月6号分区的数据出现在了9999-12-31分区，就要对齐进行修正
之后再出错日期修正的基础上，一天一天再跑用户维度的任务即可

DWD层建模

dwd建模步骤

选择业务过程，该业务过程为原子性的，比如加购的业务过程就不可拆分
确定粒度，比如下单事务事实表中，最细粒度可选为一笔订单中的一个商品
确定维度，确定业务过程的描述信息，比如下单事实，可以选择，订单来源，订单用户等维度
确定事实，关注事实的度量值，比如一次下单中，可用度量为下单金额，下单件数等事实表分类
事务事实表，一个原子的业务过程对应一张事实型事实表
周期型事实表，对于一些存量型指标，比如用户余额，商品库存，购物车存量等类型指标（各分类商品购物车存量top3）
周期同步即可
累计型事实表，常用于多事务关联指标，比如最近一日支付完成的订单从下单到支付的平均时间
多个业务过程关联，每个业务过程一般需要一个时间进行标记

数仓搭建的流程

数据调研

业务调研
同后端工程师交流，熟悉业务场景
需求调研
同产品经理讨论需求，熟悉指标，并把指标拆分为原子指标，派生指标，衍生指标
- 原子指标业务过程 + 度量值 + 聚合逻辑比如订单总额（下单，下单金额，sum）
- 派生指标统计周期 + 业务限定 + 原子指标比如最近一周各个品类的订单总额（最近一周，各品类，订单总额）
- 衍生指标统计周期，业务过程，统计粒度都相同的派生指标可以得到衍生指标，比如最近一周各个品类的退单率（最近一周各个品类的退单数 / 最近一周各个品类的下单数）

划分数据域

数仓数据域

用户域包括的事实表用户登录，用户注册
交易域加购，下单，退单，支付，退款等
流量域启动，页面，行动，曝光，错误
互动域点赞，评论，收藏
工具域下单时领用优惠券，支付时使用优惠券

构建业务总线矩阵

按照划分的数据域，和之前业务调研，确定维度，事实表，事实表度量，建立矩阵
200张事实表？？
6张维度表确定数仓架构和架构选型
业务数据：

plaintext
1. mysql -> maxwell -> kafka -> flume -> hdfs -> datahouse 
2. mysql -> datax -> hdfs

日志数据：

plaintext
1. springboot -> nginx -> 日志服务器 -> flume -> kakfa -> flume -> hdfs

数仓分层

ODS

对来自hdfs的原始数据做一个备份
对表进行分区，减少了全表扫描
对数据使用了压缩，gzip

DIM和DWD

使用hive sql对数据进行清洗

对用户数据进行了脱敏，使用MD5加密
对使用了嵌套类型的字段进行拆分，再ODS层时通过指定hive的序列化和反序列化器为`ROW FORMAT SERDE 'org.apache.
hadoop.hive.serde2.JsonSerDe' STORED AS textfile`,可以解析hdfs上的json格式数据
数据的去重，比如使用group by，开窗等方式
过滤脏数据等

使用了orc列式存储，snappy压缩（解压缩速率快，dim层和dwd层数据访问频率高）

SQL书写

测试，上线

离线项目串讲

主要思路：
数据调研
划分数据域
建立业务总线矩阵
确定数仓架构和选型
维度建模
sql编写，测试，上线

数据来源
mysql 和日志服务器，使用ezdml 梳理业务表关系
数仓建模部分
数仓架构和选型数仓架构流程图

30个数仓指标

指标名	数据域	解释
各个引流渠道的跳出率	流量域	只有一个页面的会话 / 该渠道的总会话数
路径分析	流量域	页面跳转统计
注册漏斗分析	流量域	启动app，浏览页面，注册结果，提交订单，支付订单
关键词搜索频次分布	流量域	用户输入关键词搜索后返回结果统计，用于调整用户搜索体验
商品复购率	商品域	重复购买次数 / 总购买次数
最近1，7，30日的下单人数，订单数	商品域
各品类购车车存量top3	商品域
商品收藏量top3	商品域
下单成功的订单中，下单到支付的平均时间间隔	交易域
线上会员订单业务类型分布	交易域	各个注册来源的会员在支付时支付方式统计，比如支付包，微信，收银机等
各省份GMV（交易总额）	交易域
客单价	交易域	下单总额 / 下单人数
购买频次分布	交易域	当日支付1次的人数，支付2次的人数...
会员客单价	交易域	会员的总支付金额 / 总会员人数
会员消费频次	交易域
会员复购分析	交易域
消费时段偏好	交易域
各个优惠券使用情况，比如领取，下单，支付	工具域
7日新增和活跃统计	用户域
用户的留存分析	用户域
用户流失统计	用户域
注册用户LTV价值分析	用户域	注册用户随着时间流逝，每天支付总额分析
用户漏斗分析	用户域
新增下单用户统计	用户域
会员身份分布	用户域
会员来源统计	用户域

即席查询工具 - prestoSQL（trino）

trino

数据治理

元数据管理 atlas

apache atlas官网

数据质量监控

自己写SQL维护，并结合后端工程师的springboot项目，提供置空率，指标波动等告警

08离线数仓

离线数仓

离线数仓分层的原因和背景

数仓的数据来源和数据产出

常见数仓组件的版本

集群云服务器

数仓建模

Dim建模

dim建模步骤

2.2 dim_user_info 实例

DWD层建模

dwd建模步骤

数仓搭建的流程

数据调研

划分数据域

构建业务总线矩阵

数仓分层

ODS

DIM和DWD

SQL书写

测试，上线

离线项目串讲

30个数仓指标

即席查询工具 - prestoSQL（trino）

数据治理

元数据管理 atlas

数据质量监控

权限管理 apache ranger

数据中台（没做）

08离线数仓