现代数据仓库的建设之路

2022-04-18

什么是数据仓库

数据仓库，由数据仓库之父比尔·恩门（Bill Inmon）于1990年提出，他在1991年出版的“Building the Data Warehouse”（《建立数据仓库》）一书中所提出的数据仓库的定义——数据仓库（Data Warehouse）是一个面向主题的（Subject Oriented）、集成的（Integrated）、相对稳定的（Non-Volatile）、反映历史变化（Time Variant）的数据集合，用于支持管理决策(Decision Making Support)。
现在我们构建的数据仓库，仍然是在Bill Inmon提出的数仓理论基础之上，只不过现在根据时代不同，需求变了，衍生出各种工具和技术来支持数据仓库的发展。比如：大数据技术，数据湖技术，流计算技术等。

什么是现代数仓

现代数仓是大数据技术以及人工智能技术的发展基础上而出现的数仓架构技术。其实就是随着企业信息化、互联网技术的发展，数据量变的越来越大，数据格式越来越多，决策要求越来越苛刻，而出现的针对行解决方案。

现代数仓的技术架构

我们先来说说传统的数仓技术。其实数据仓库很早之前就有了，也就是说在现代数仓之前，有很多传统的数仓技术，例如基于teradata，ibm db2的数据仓库，只不过是数据仓库技术在大数据背景下发生了很多改变，也就是我们开始抛弃了传统构建数仓的技术，转而选择了更能满足当前时代需求的大数据技术而已，当然大数据技术并没有完整的、彻底的取代传统的技术实现，我们依然可以在很多地方看见它们的身影

所以，以大数据技术为代表的数仓建设，构成了现代数仓的基础技术架构。如下图所示：

图中展示了现代数仓基本技术架构方案。

数仓建设的方法

数仓建设的核心就是数仓的分层模型的建设。数仓的分层模型建设，主要可分为如下几个方面的建设：

主题域划分规范
分层划分规范
命名规范

主题域的规范

数据域划分过程

主要是按照业务过程进行主题域划分，一般分为一级主题和二级主题

比如：交易域，我们总共分为了6个二级主题，分别为：加购、下单、退款、换单、营销、物流

命名规范

命名规范的确定对于数仓建设来说，是非常重要的。良好的命名规范不仅可以提高数仓的开发效率，减少沟通成本，而且对于数据治理，减少数据出错，提高数据质量非常有帮助。所以，在我长期实践中来说，建设一个数仓从命名规范开始很有必要。

层级域	命名规范	备注
ods层	ods.ods_库表_表名_更新周期(day/hour/week/month)_(da或者di)	ods_db_trade_order_tb_order_di
mid层	库名.mid_库表_表名_更新周期_(da或者di)	此层可选，不是必须有，但是涉及到多业务系统融合的时候需要此层。如：mid_db_order_tb_order
mds层	库名.mds_一级主题简称[_二级主题简称]_自定义表名_更新周期_(da或者di)	mds_sales_order_product_detail
eds层	库名.eds_(topic/aggr)_一级主题简称[_二级主题简称]_自定义表名_更新周期_(da或者di)
ads层	库名.ads_一级主题简称[_二级主题简称]_自定义表名_更新周期_(da或者di)

每层按照规定的前缀命名开始：ods,mid,mds,eds,ads，中间加上一二级主题
命名，最后确定数据周期以及数据是增量还是全量。

day: 数据按天更新
hour: 数据按小时更新
week: 数据按周更新
month: 数据按月更新
year: 数据按年更新
da: 全量数据
di: 增量数据

分层规范

如上图所示：灰色框是提出的分层模式，mds层分为原子层与轻度汇总层，对使用者开放权限优先级eds->mds汇总层
详细说明如下：

ETL任务设计和开发

主要包括2种任务的开发

离线hive任务开发
实时flink 任务开发
离线hive任务开发
对于T+1的数据来说，我们主要使用hive sql 进行etl任务开发。ods层数据我们主要通过统一数据同步工具进行数据接入（通过包装sqoop工具进行，增量或者全量数据接入）。对于数据量超过千万的数据，进行增量接入。

mds层使用hive sql 进行etl操作后最终写入到hive数仓中，如下所示：

set hive.auto.convert.join=false;
set hive.exec.parallel=true;
set hive.exec.parallel.thread.number=10;
set hive.exec.reducers.bytes.per.reducer=512000000;
use ods;

DROP TABLE IF EXISTS mds.temp1;
CREATE TEMPORARY TABLE mds.temp1 AS
select 
    a.order_id      --          COMMENT '订单ID'
    ,a.order_source_code     --          COMMENT '订单来源代码'
    ,a.order_no     --          COMMENT '订单编号'
    ,a.product_id     --          COMMENT '产品编号'
    ,a.product_name   --          COMMENT '产品名称'
    ,a.user_id        --          COMMENT '用户编号'
    ,a.user_name      --          COMMENT '用户名称'
    ,a.order_type_code  --               COMMENT '订单类型代码1购买、2换课购买、3退课、4换课退课'
    ,a.order_state_code  --               COMMENT '订单状态代码0未支付、1已支付、2取消'
    ,a.order_product_state_code    --           COMMENT '订单产品状态代码0：未支付,1：已支付,2:取消.3:冻结.4:退课中.5:已退课.6:过期.7:休学.8:换课中.9:已换课'
    ,a.order_time     --          COMMENT '下单时间'
    ,a.order_pay_time     --          COMMENT '订单支付时间'
    ,a.order_cancel_type_code     --          COMMENT '订单取消类型代码0：用户操作，1：系跑批取消，2：客服取消'
    ,a.order_cancel_time       --        COMMENT '订单取消时间'
    ,a.ip       --        COMMENT '产生订单IP'
    ,a.order_fputype_code     --          COMMENT '下单终端类型代码'
    ,a.order_scene_code      --         COMMENT '下单场景类型代码结算(0),领取(1),代理商(2),天猫开单(3),第三方订单(4),校代订单(5),快捷下单(6), 换课(7),虚拟卡(8),团购(9),小程序购课(10)'
    ,a.order_channel_code    --           COMMENT '下单渠道类型代码0:官网.1:网盟.2:校代.3:天猫. 4:虚拟代理商.5:返点代理商'
    ,a.give_product_sign     --          COMMENT '赠送产品标志1是 0否'
    ,a.finance_type_code     --          COMMENT '财务核算类型代码0 老版本，1 新版本'
    ,a.basic_order_sign      --       COMMENT '原订单标志1是 0否'
    ,a.basic_order_id        --       COMMENT '原订单ID'
    ,a.agent_order_sign      --         COMMENT '代理商订单标志1是 0否'
    ,a.agent_order_id        --       COMMENT '代理商ID'
    ,a.agent_account_id      --         COMMENT '代理商账户ID'
    ,a.agent_account_ver_id   --            COMMENT '代理商账户版本ID'
    ,a.un_sign        --       COMMENT '网盟标志1是 0否'
    ,a.aid     --          COMMENT 'AID'
    ,a.kid     --          COMMENT 'KID'
    ,a.last_aid     --          COMMENT '最后一次点击AID'
    ,a.last_kid     --          COMMENT '最后一次点击KID'
    ,b.discount_price   --           COMMENT '成交价格'
    ,case when a.basic_order_sign=0 then -1*b.net_value else b.net_value end net_value --         COMMENT '净值'
    ,c.mix_gold_content --                COMMENT '综合含金量'
    ,b.price --原价
    ,b.promote_after_amount --优惠券后价格
    ,a.on_off_line_code -- COMMENT '线上线下类型代码1:线下 2：线上'
    ,a.on_off_line_name -- COMMENT'线上线下类型代码名称'
    ,a.campus_id -- COMMENT '校代ID'
    ,a.put_source_channel -- COMMENT '投放来源渠道'
    ,a.put_url -- COMMENT '投放URL'
    ,a.order_channel_v2_code
    ,a.over_time   -- 过期时间
    ,a.freeze_time  -- 冻结时间
    ,a.create_time  -- 开课时间
    ,a.seller_user_id --销售用户id
    ,a.seller_domain_account --销售域账号
    ,a.seller_real_name --销售真实姓名
    ,a.seller_id --销售id
    ,greatwall_order_sign  --长城订单标志
    ,a.resubmit_path
    ,a.sub_order_type_code
    ,case when a.basic_order_sign=0 then -1*nvl(d.check_value,b.net_value) else nvl(d.check_value,b.net_value) end check_value --         COMMENT '业绩净值'
from
(
    select
        order_id      --          COMMENT '订单ID'
        ,order_source_code     --          COMMENT '订单来源代码'
        ,order_no     --          COMMENT '订单编号'
        ,product_id     --          COMMENT '产品编号'
        ,product_name   --          COMMENT '产品名称'
        ,user_id        --          COMMENT '用户编号'
        ,user_name      --          COMMENT '用户名称'
        ,order_type_code  --               COMMENT '订单类型代码1购买、2换课购买、3退课、4换课退课'
        ,order_state_code  --               COMMENT '订单状态代码0未支付、1已支付、2取消'
        ,order_product_state_code    --           COMMENT '订单产品状态代码0：未支付,1：已支付,2:取消.3:冻结.4:退课中.5:已退课.6:过期.7:休学.8:换课中.9:已换课'
        ,order_time     --          COMMENT '下单时间'
        ,order_pay_time     --          COMMENT '订单支付时间'
        ,order_cancel_type_code     --          COMMENT '订单取消类型代码0：用户操作，1：系跑批取消，2：客服取消'
        ,order_cancel_time       --        COMMENT '订单取消时间'
        ,ip       --        COMMENT '产生订单IP'
        ,order_fputype_code     --          COMMENT '下单终端类型代码'
        ,order_scene_code      --         COMMENT '下单场景类型代码结算(0),领取(1),代理商(2),天猫开单(3),第三方订单(4),校代订单(5),快捷下单(6), 换课(7),虚拟卡(8),团购(9),小程序购课(10)'
        ,order_channel_code    --           COMMENT '下单渠道类型代码0:官网.1:网盟.2:校代.3:天猫. 4:虚拟代理商.5:返点代理商'
        ,give_product_sign     --          COMMENT '赠送产品标志1是 0否'
        ,finance_type_code     --          COMMENT '财务核算类型代码0 老版本，1 新版本'
        ,basic_order_sign      --       COMMENT '原订单标志1是 0否'
        ,basic_order_id        --       COMMENT '原订单ID'
        ,agent_order_sign      --         COMMENT '代理商订单标志1是 0否'
        ,agent_order_id        --       COMMENT '代理商ID'
        ,agent_account_id      --         COMMENT '代理商账户ID'
        ,agent_account_ver_id   --            COMMENT '代理商账户版本ID'
        ,un_sign        --       COMMENT '网盟标志1是 0否'
        ,aid     --          COMMENT 'AID'
        ,kid     --          COMMENT 'KID'
        ,last_aid     --          COMMENT '最后一次点击AID'
        ,last_kid     --          COMMENT '最后一次点击KID'
        --,sum(discount_price)  discount_price   --           COMMENT '成交价格'
        --,sum(net_value)     net_value  --         COMMENT '净值'
        --,sum(mix_gold_content)  mix_gold_content --                COMMENT '综合含金量'
        ,on_off_line_code -- COMMENT '线上线下类型代码1:线下 2：线上'
        ,on_off_line_name -- COMMENT'线上线下类型代码名称'
        ,campus_id -- COMMENT '校代ID'
        ,put_source_channel -- COMMENT '投放来源渠道'
        ,put_url -- COMMENT '投放URL'
        ,order_channel_v2_code
        ,over_time   -- 过期时间
        ,freeze_time  -- 冻结时间
        ,create_time  -- 开课时间
        ,seller_user_id --销售用户id
        ,seller_domain_account --销售域账号
        ,seller_real_name --销售真实姓名
        ,seller_id --销售id
        ,greatwall_order_sign  --长城订单标志
        ,resubmit_path
        ,sub_order_type_code
    from mds.mds_sales_order_product_pay_detail
    where dt='${hivevar:dt}'
    group by order_id      --          COMMENT '订单ID'
        ,order_source_code     --          COMMENT '订单来源代码'
        ,order_no     --          COMMENT '订单编号'
        ,product_id     --          COMMENT '产品编号'
        ,product_name   --          COMMENT '产品名称'
        ,user_id        --          COMMENT '用户编号'
        ,user_name      --          COMMENT '用户名称'
        ,order_type_code  --               COMMENT '订单类型代码1购买、2换课购买、3退课、4换课退课'
        ,order_state_code  --               COMMENT '订单状态代码0未支付、1已支付、2取消'
        ,order_product_state_code    --           COMMENT '订单产品状态代码0：未支付,1：已支付,2:取消.3:冻结.4:退课中.5:已退课.6:过期.7:休学.8:换课中.9:已换课'
        ,order_time     --          COMMENT '下单时间'
        ,order_pay_time     --          COMMENT '订单支付时间'
        ,order_cancel_type_code     --          COMMENT '订单取消类型代码0：用户操作，1：系跑批取消，2：客服取消'
        ,order_cancel_time       --        COMMENT '订单取消时间'
        ,ip       --        COMMENT '产生订单IP'
        ,order_fputype_code     --          COMMENT '下单终端类型代码'
        ,order_scene_code      --         COMMENT '下单场景类型代码结算(0),领取(1),代理商(2),天猫开单(3),第三方订单(4),校代订单(5),快捷下单(6), 换课(7),虚拟卡(8),团购(9),小程序购课(10)'
        ,order_channel_code    --           COMMENT '下单渠道类型代码0:官网.1:网盟.2:校代.3:天猫. 4:虚拟代理商.5:返点代理商'
        ,give_product_sign     --          COMMENT '赠送产品标志1是 0否'
        ,finance_type_code     --          COMMENT '财务核算类型代码0 老版本，1 新版本'
        ,basic_order_sign      --       COMMENT '原订单标志1是 0否'
        ,basic_order_id        --       COMMENT '原订单ID'
        ,agent_order_sign      --         COMMENT '代理商订单标志1是 0否'
        ,agent_order_id        --       COMMENT '代理商ID'
        ,agent_account_id      --         COMMENT '代理商账户ID'
        ,agent_account_ver_id   --            COMMENT '代理商账户版本ID'
        ,un_sign        --       COMMENT '网盟标志1是 0否'
        ,aid     --          COMMENT 'AID'
        ,kid     --          COMMENT 'KID'
        ,last_aid     --          COMMENT '最后一次点击AID'
        ,last_kid     --          COMMENT '最后一次点击KID'
        ,on_off_line_code -- COMMENT '线上线下类型代码1:线下 2：线上'
        ,on_off_line_name -- COMMENT'线上线下类型代码名称'
        ,campus_id -- COMMENT '校代ID'
        ,put_source_channel -- COMMENT '投放来源渠道'
        ,put_url -- COMMENT '投放URL'
        ,order_channel_v2_code
        ,over_time   -- 过期时间
        ,freeze_time  -- 冻结时间
        ,create_time  -- 开课时间
        ,seller_user_id --销售用户id
        ,seller_domain_account --销售域账号
        ,seller_real_name --销售真实姓名
        ,seller_id --销售id
        ,greatwall_order_sign  --长城订单标志
        ,resubmit_path
        ,sub_order_type_code
)a
left join  
(
    select  discount_price,net_value,price,promote_after_amount,order_no,product_id
    from mid.mid_db_order_tb_order_product
    where dt='${hivevar:dt}'
    group by  price,promote_after_amount,order_no,product_id,discount_price,net_value
)b
on  a.order_no=b.order_no
and a.product_id=b.product_id
left join
(
    select
    order_no
    ,mix_gold_content
    from
    mid.mid_db_order_tb_order
    where dt='${hivevar:dt}'
)c 
on a.order_no = c.order_no
left join 
(-- 此关联多余, 主表的check_value就是该表关联出来
    select 
    order_no 
    ,product_id
    ,performance_value check_value
    from 
    ods.ods_db_order_pe_order_product_performance
    where dt = '${hivevar:dt}'
    and is_deleted = 0
) d
on  a.order_no=d.order_no
and a.product_id=d.product_id
;




-- 插入物理表
INSERT OVERWRITE TABLE mds.mds_sales_order_product_detail PARTITION (dt='${hivevar:dt}')
select
order_id      --          COMMENT '订单ID'
,order_source_code     --          COMMENT '订单来源代码'
,order_no     --          COMMENT '订单编号'
,product_id     --          COMMENT '产品编号'
,product_name   --          COMMENT '产品名称'
,user_id        --          COMMENT '用户编号'
,user_name      --          COMMENT '用户名称'
,order_type_code  --               COMMENT '订单类型代码1购买、2换课购买、3退课、4换课退课'
,order_state_code  --               COMMENT '订单状态代码0未支付、1已支付、2取消'
,order_product_state_code    --           COMMENT '订单产品状态代码0：未支付,1：已支付,2:取消.3:冻结.4:退课中.5:已退课.6:过期.7:休学.8:换课中.9:已换课'
,order_time     --          COMMENT '下单时间'
,order_pay_time     --          COMMENT '订单支付时间'
,order_cancel_type_code     --          COMMENT '订单取消类型代码0：用户操作，1：系跑批取消，2：客服取消'
,order_cancel_time       --        COMMENT '订单取消时间'
,ip       --        COMMENT '产生订单IP'
,order_fputype_code     --          COMMENT '下单终端类型代码'
,order_scene_code      --         COMMENT '下单场景类型代码结算(0),领取(1),代理商(2),天猫开单(3),第三方订单(4),校代订单(5),快捷下单(6), 换课(7),虚拟卡(8),团购(9),小程序购课(10)'
,order_channel_code    --           COMMENT '下单渠道类型代码0:官网.1:网盟.2:校代.3:天猫. 4:虚拟代理商.5:返点代理商'
,give_product_sign     --          COMMENT '赠送产品标志1是 0否'
,finance_type_code     --          COMMENT '财务核算类型代码0 老版本，1 新版本'
,basic_order_sign      --       COMMENT '原订单标志1是 0否'
,basic_order_id        --       COMMENT '原订单ID'
,agent_order_sign      --         COMMENT '代理商订单标志1是 0否'
,agent_order_id        --       COMMENT '代理商ID'
,agent_account_id      --         COMMENT '代理商账户ID'
,agent_account_ver_id   --            COMMENT '代理商账户版本ID'
,un_sign        --       COMMENT '网盟标志1是 0否'
,aid     --          COMMENT 'AID'
,kid     --          COMMENT 'KID'
,last_aid     --          COMMENT '最后一次点击AID'
,last_kid     --          COMMENT '最后一次点击KID'
,discount_price   --           COMMENT '成交价格'
,net_value --         COMMENT '净值'
,mix_gold_content --                COMMENT '综合含金量'
,price --原价
,promote_after_amount --优惠券后价格 
,on_off_line_code -- COMMENT '线上线下类型代码1:线下 2：线上'
,on_off_line_name -- COMMENT'线上线下类型代码名称'
,campus_id -- COMMENT '校代ID'
,put_source_channel -- COMMENT '投放来源渠道'
,put_url -- COMMENT '投放URL'
,order_channel_v2_code
,over_time   -- 过期时间
,freeze_time  -- 冻结时间
,create_time  -- 开课时间
,seller_user_id --销售用户id
,seller_domain_account --销售域账号
,seller_real_name --销售真实姓名
,seller_id --销售id
,greatwall_order_sign  --长城订单标志
,resubmit_path
,sub_order_type_code
,check_value
from
mds.temp1;

mds层往往代码是非常复杂的，关联表多，字段多，业务复杂，所以这层是建设的重点。

对于其它层来说，开发过程类似，主要是熟悉业务过程，理解业务，开发代码。
任务开发完成后，需要进行上线操作，主要确定任务依赖的上游任务，任务运行周期，报警配置等操作，这些完成后，通过审批后，任务都上线完成。

实时flink/doris任务开发

对于T+1的任务来说，数据时效性可能不满足需求，这时我们会开发实时flink和doris任务来满足这类需求。

对于，秒级数据延迟需求的任务，我们会直接开发flink任务。整个数据处理链路如下：

数据源：通过canal接入业务binlog数据到kafka中
数据加工：通过flink sql来实现数据业务逻辑，最后输出到doris中
数据服务：通过数据API或者报表方式提供数据服务

如下是一个flink sql 例子：通过消费kafka里的线索数据，通过flink sql结合python udf(算法模型)，计算出线索分，最后输出到rocketmq 供业务方消费使用

----------------------------------------------------
-- Flink SQL Start
-- Flink SQL Desc 实时线索分计算
-- Author:
-- Update:
------------------------------------------------------
-- 创建源表
-- use catalog default_catalog;

set table.sql-dialect=default;
-- 创建 rocketmq - kafka 源表
CREATE TABLE `source_bdp_cluescore_realtime_topic` (
    `clueId` INT  COMMENT '线索ID',
    `cluePoolCode` INT  COMMENT '线索池ID',
    `ssoId` INT  COMMENT '学员SSOId',
    `city` STRING  COMMENT '城市编码',
    `firstSource` STRING  COMMENT '一级来源名称',
    `secondSource` STRING  COMMENT '二级来源名称',
    `scene` STRING  COMMENT 'scene值',
    `departmentTypeName` STRING  COMMENT '渠道部门名称',
    `registerTime` STRING  COMMENT '注册时间'
) with (
'connector' = 'kafka',
'topic' = 'bdp-cluescore-realtime-topic',
'properties.bootstrap.servers' = 'kafkaXXX',
'scan.startup.mode'='earliest-offset',
--'scan.startup.mode'='timestamp',
--'scan.startup.timestamp-millis'='1623030860000',
'properties.group.id'='bdp-cluescore-realtime-group',
'properties.security.protocol'='SASL_PLAINTEXT',
'properties.sasl.mechanism'='PLAIN',
'properties.sasl.jaas.config'='org.apache.kafka.common.security.plain.PlainLoginModule required username="xxx" password="xxx";',
'format' = 'json',
'json.fail-on-missing-field' = 'false',
'json.ignore-parse-errors' = 'true'
);


-- 创建目标 rocketmq表在flink内部的的映射关系
CREATE TABLE `sink_rocket_realtime_cluescore` (
    `clue_id` STRING  COMMENT '线索ID',
    `user_id` STRING  COMMENT '学员SSOId',
    `clue_pool_code` STRING  COMMENT '线索池ID',
    `score` DECIMAL(8,2)  COMMENT '线索分',
    `data_version` STRING  COMMENT '数据版本',
    `dt` STRING  COMMENT '数据计算日期'
) with (
    'connector' = 'koolearn-rocketmq',
    'topic' = 'bigdata_sync_table_ads_clue_layer_score_topic',
    'nameserver' = 'xxx',
    'group'='realtime-clue-score-producer',
    'tag'='ads_clue_layer_score',
    'product_send_async'='false',
    'product_send_batch'='3000',
    'product_send_interval' = '1000',
    'format' = 'json'
);

insert into sink_rocket_realtime_cluescore
-- insert into test_print
select
cast(clueId as string) as clue_id ,
cast(ssoId as string) as user_id,
cast(cluePoolCode as string) as clue_pool_code,
calc_clue_score(cast(ssoId as string),city,firstSource,secondSource,departmentTypeName,registerTime) as score,
'v4' as data_version,
FROM_UNIXTIME(UNIX_TIMESTAMP(),'yyyyMMdd') as dt
 from source_bdp_cluescore_realtime_topic 
 where cluePoolCode in (9,11)
 ;

对于，分钟级数据需求来说，我们主要使用doris数据库，通过微批方式进行数据处理。数据处理流程如下所示：

上图是目前我们采用的架构方案，总体流程如下：

数据接入部分，分为业务数据和日志数据。业务数据通过binlog方式收集到kafka后，在通过flink写入到doris ods层中
mds层，采用每10分钟、半小时、一小时进行增量或全量的方式更新，构建业务模型层
ads层，构建大宽表层加速上层查询速度
对于一些分析类查询需求，通过doris的export功能导出到hive通过presto提供查询
BI查询直接通过mysql协议访问db，配合查询层缓存来提供报表分析服务
每层ETL任务通过自研调度系统调度运行，报警监控一体化