公司实战案例

MySQL到数仓

创建 MySQL 原始数据得视图层

drop view if exists dwd.mysqldbname_mysqltablename;
create view if not exists dwd.mysqldbname_mysqltablename
as select *
 from ods_mysqldbname.mysqltablename

1. ods_mysqldbname 中 mysqldbname 表示的是 MySQL的 mysqldbname库，mysqltablename 表示 MySQL 的表。
2. 先用同步工具把 MySQL 中的 mysqldbname.mysqltablename 表同步到 Hive。
3. 创建一个 dwd.mysqldbname_mysqltablename 的 ods视图。

日志数据

create external table if not EXISTS kafka.aliyun_bigdata_message_push_log
(
    jsondata string
)
comment '消息推送日志'
partitioned by (dates string)
location 'hdfs://bigdata-cdh.ld-isilon.bigdata:8020/data/kafka/aliyun_bigdata_message_push_log';
alter table kafka.aliyun_bigdata_message_push_log drop if exists partition(dates={today,yyyyMMdd});
alter table kafka.aliyun_bigdata_message_push_log add if not exists partition (dates={today,yyyyMMdd}) location 'hdfs://bigdata-cdh.ld-isilon.bigdata:8020/data/kafka/aliyun_bigdata_message_push_log/{today,yyyyMMdd}';

ODS 层

建立原始 ODS 表

CREATE
EXTERNAL TABLE IF NOT
EXISTS applydata_bi_ods.ods_mysqldbname_mysqltablename 
    ( 
    id int COMMENT "PK" ,plan_name string COMMENT "方案名称" ,
    portrait_type string COMMENT "人群类型 1，画像人群；2，单个用户；3，批量用户；" ,
    portrait_id string COMMENT "人群ID" ,
    predict_number int COMMENT "预估人数" ,
    push_category int COMMENT "推送类别 1，会员福利；2，系统通知；" ,
    push_type string COMMENT "推送方式，逗号分割 1，push推送；2，短信推送" ,
 ) PARTITIONED BY (dates int) STORED AS ORC;

ODS 只存储一年的数据

ALTER TABLE applydata_bi_ods.ods_mysqldbname_mysqltablename SET TBLPROPERTIES ('EXTERNAL' = 'FALSE');
ALTER TABLE applydata_bi_ods.ods_mysqldbname_mysqltablename
    DROP IF EXISTS PARTITION (dates < {today-365,yyyyMMdd});
ALTER TABLE applydata_bi_ods.ods_mysqldbname_mysqltablename SET TBLPROPERTIES ('EXTERNAL' = 'TRUE');

导入 T+1 的数据

INSERT OVERWRITE TABLE applydata_bi_ods.ods_mysqldbname_mysqltablename PARTITION (dates={today-1, yyyyMMdd})
SELECT id, plan_name, portrait_type, portrait_id, predict_number, push_category, push_type, client_notice, message_store, title, content, sms_channel, sms_content, link_type, link, coupon_code, import_status, status, import_number, push_success_number, sms_success_number, reason, import_start_time, import_end_time, push_start_time, push_end_time, oper_name, create_time, update_time
FROM dwd.mysqldbname_mysqltablename;

1. 把 MySQL 倒过来的原始数据写入到ods层，也就是 MySQL 的每天数据存储到数仓的 ODS 层，并且保留一年的数据。

DWD 层

建表

CREATE EXTERNAL
TABLE IF NOT EXISTS applydata_bi_dw.dwd_mysqldbname_mysqltablename (
  id int COMMENT "PK",
  plan_name string COMMENT "方案名称",
) PARTITIONED BY (dates int) STORED AS ORC;

控制分区数

ALTER TABLE applydata_bi_dw.dwd_mysqldbname_mysqltablename SET TBLPROPERTIES ('EXTERNAL' = 'FALSE');
ALTER TABLE applydata_bi_dw.dwd_mysqldbname_mysqltablename DROP IF EXISTS PARTITION(dates < {today-90,yyyyMMdd});
ALTER TABLE applydata_bi_dw.dwd_mysqldbname_mysqltablename SET TBLPROPERTIES ('EXTERNAL' = 'TRUE');

写入数据

DWD 层，对于 ODS 层做了轻度的清洗层，比如时间格式化，和一些表的关联操作。

INSERT OVERWRITE TABLE applydata_bi_dw.dwd_mysqldbname_mysqltablename PARTITION(dates={today-1,yyyyMMdd})
SELECT
  id,
  plan_name,
FROM
  applydata_bi_ods.ods_mysqldbname_mysqltablename
WHERE
  dates={today-1,yyyyMMdd};

DWS

按天进行计算 (派生指标)

ADS

衍生指标

分桶表

CREATE EXTERNAL TABLE IF NOT EXISTS applydata_bi_dwl.dwl_cc_label_bucket(
  bucket_key string COMMENT '分桶键',
  value string COMMENT '标签值', 
  `timestamp` int COMMENT '更新时间戳')
COMMENT '标签分桶表'
PARTITIONED BY (dates int)
CLUSTERED BY (bucket_key) SORTED BY(bucket_key) INTO 500 BUCKETS
STORED AS ORC
;

ALTER TABLE applydata_bi_dwl.dwl_cc_label_bucket SET TBLPROPERTIES ('EXTERNAL' = 'FALSE');
ALTER TABLE applydata_bi_dwl.dwl_cc_label_bucket DROP IF EXISTS PARTITION(dates < {today-31,yyyyMMdd});
ALTER TABLE applydata_bi_dwl.dwl_cc_label_bucket SET TBLPROPERTIES ('EXTERNAL' = 'TRUE');

SET hive.enforce.bucketing=true;
SET hive.enforce.sorting = true;
SET hive.auto.convert.join = true;
SET hive.enforce.bucketing=true;
SET hive.auto.convert.join=false;
SET hive.auto.convert.sortmerge.join=true;
SET hive.optimize.bucketmapjoin=true;
SET hive.optimize.bucketmapjoin.sortedmerge=true;
SET hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;

INSERT OVERWRITE TABLE applydata_bi_dwl.dwl_cc_label_bucket PARTITION(dates = {today-1,yyyyMMdd})
SELECT
     CONCAT(t1.id_type,',',t1.id,',',t1.label_id) AS bucket_key
    ,t1.value
    ,t1.`timestamp`
FROM
    applydata_bi_dwl.dwl_cc_label t1
INNER JOIN
    applydata_bi_dim.dim_label t2
ON t1.label_id = t2.label_id
WHERE
    t1.dates = {today-1,yyyyMMdd}
    AND t2.dates = {today-1,yyyyMMdd}
    AND t2.status = 1 
    AND t2.is_delete = 0
    AND t2.is_real_leaf = 1
;