离线数仓 (十三) --------- DWD 层搭建

深碍√TFBOYSˉ_ 2024-04-01 11:30 153阅读 0赞

目录

  • 前言
  • 一、DWD 层 (用户行为日志)
      1. 日志解析思路
      1. get_json_object 函数使用
      1. 启动日志表
      1. 页面日志表
      1. 动作日志表
      1. 曝光日志表
      1. 错误日志表
      1. DWD 层用户行为数据加载脚本
  • 二、DWD层 (业务数据)
      1. 评价事实表 (事务型事实表)
      1. 订单明细事实表 (事务型事实表)
      1. 退单事实表 (事务型事实表)
      1. 加购事实表 (周期型快照事实表，每日快照)
      1. 收藏事实表 (周期型快照事实表，每日快照)
      1. 优惠券领用事实表 (累积型快照事实表)
      1. 支付事实表 (累积型快照事实表)
      1. 退款事实表 (累积型快照事实表)
      1. 订单事实表 (累积型快照事实表)
      1. DWD 层业务数据首日装载脚本
      1. DWD层业务数据每日装载脚本

前言

DWD 层主要完成的功能 ：

  • 对用户行为数据解析。
  • 对业务数据采用维度模型重新建模。

一、DWD 层 (用户行为日志)

1. 日志解析思路

A、日志结构回顾

页面埋点日志

在这里插入图片描述
启动日志在这里插入图片描述
B、日志解析思路

在这里插入图片描述

2. get_json_object 函数使用

A、数据

  1. [{
  2.     "name":"大郎","sex":"男","age":"25"},{
  3.     "name":"西门庆","sex":"男","age":"47"}]

B、取出第一个 json 对象

  1. hive (gmall)>
  2. select get_json_object('[{"name":"大郎","sex":"男","age":"25"},{"name":"西门庆","sex":"男","age":"47"}]','$[0]');

结果是：{"name":"大郎","sex":"男","age":"25"}

C、取出第一个 json 的 age 字段的值

  1. hive (gmall)>
  2. SELECT get_json_object('[{"name":"大郎","sex":"男","age":"25"},{"name":"西门庆","sex":"男","age":"47"}]',"$[0].age");

结果是：25

3. 启动日志表

启动日志解析思路：启动日志表中每行数据对应一个启动记录，一个启动记录应该包含日志中的公共信息和启动信息。先将所有包含start字段的日志过滤出来，然后使用get_json_object函数解析每个字段。

在这里插入图片描述

A、建表语句

  1. DROP TABLE IF EXISTS dwd_start_log;
  2. CREATE EXTERNAL TABLE dwd_start_log(
  3.     `area_code` STRING COMMENT '地区编码',
  4.     `brand` STRING COMMENT '手机品牌',
  5.     `channel` STRING COMMENT '渠道',
  6.     `is_new` STRING COMMENT '是否首次启动',
  7.     `model` STRING COMMENT '手机型号',
  8.     `mid_id` STRING COMMENT '设备id',
  9.     `os` STRING COMMENT '操作系统',
  10.     `user_id` STRING COMMENT '会员id',
  11.     `version_code` STRING COMMENT 'app版本号',
  12.     `entry` STRING COMMENT 'icon手机图标 notice 通知 install 安装后启动',
  13.     `loading_time` BIGINT COMMENT '启动加载时间',
  14.     `open_ad_id` STRING COMMENT '广告页ID ',
  15.     `open_ad_ms` BIGINT COMMENT '广告总共播放时间',
  16.     `open_ad_skip_ms` BIGINT COMMENT '用户跳过广告时点',
  17.     `ts` BIGINT COMMENT '时间'
  18. ) COMMENT '启动日志表'
  19. PARTITIONED BY (`dt` STRING) -- 按照时间创建分区
  20. STORED AS PARQUET -- 采用parquet列式存储
  21. LOCATION '/warehouse/gmall/dwd/dwd_start_log' -- 指定在HDFS上存储位置
  22. TBLPROPERTIES('parquet.compression'='lzo') -- 采用LZO压缩
  23. ;

B、数据导入

在这里插入图片描述

  1. hive (gmall)> 
  2. insert overwrite table dwd_start_log partition(dt='2020-06-14')
  3. select
  4.     get_json_object(line,'$.common.ar'),
  5.     get_json_object(line,'$.common.ba'),
  6.     get_json_object(line,'$.common.ch'),
  7.     get_json_object(line,'$.common.is_new'),
  8.     get_json_object(line,'$.common.md'),
  9.     get_json_object(line,'$.common.mid'),
  10.     get_json_object(line,'$.common.os'),
  11.     get_json_object(line,'$.common.uid'),
  12.     get_json_object(line,'$.common.vc'),
  13.     get_json_object(line,'$.start.entry'),
  14.     get_json_object(line,'$.start.loading_time'),
  15.     get_json_object(line,'$.start.open_ad_id'),
  16.     get_json_object(line,'$.start.open_ad_ms'),
  17.     get_json_object(line,'$.start.open_ad_skip_ms'),
  18.     get_json_object(line,'$.ts')
  19. from ods_log
  20. where dt='2020-06-14'
  21. and get_json_object(line,'$.start') is not null;

C、查看数据

  1. hive (gmall)> 
  2. select * from dwd_start_log where dt='2020-06-14' limit 2;

4. 页面日志表

页面日志解析思路：页面日志表中每行数据对应一个页面访问记录，一个页面访问记录应该包含日志中的公共信息和页面信息。先将所有包含 page 字段的日志过滤出来，然后使用 get_json_object 函数解析每个字段。

页面日志表解析：

在这里插入图片描述

A、建表语句

  1. DROP TABLE IF EXISTS dwd_page_log;
  2. CREATE EXTERNAL TABLE dwd_page_log(
  3.     `area_code` STRING COMMENT '地区编码',
  4.     `brand` STRING COMMENT '手机品牌',
  5.     `channel` STRING COMMENT '渠道',
  6.     `is_new` STRING COMMENT '是否首次启动',
  7.     `model` STRING COMMENT '手机型号',
  8.     `mid_id` STRING COMMENT '设备id',
  9.     `os` STRING COMMENT '操作系统',
  10.     `user_id` STRING COMMENT '会员id',
  11.     `version_code` STRING COMMENT 'app版本号',
  12.     `during_time` BIGINT COMMENT '持续时间毫秒',
  13.     `page_item` STRING COMMENT '目标id ',
  14.     `page_item_type` STRING COMMENT '目标类型',
  15.     `last_page_id` STRING COMMENT '上页类型',
  16.     `page_id` STRING COMMENT '页面ID ',
  17.     `source_type` STRING COMMENT '来源类型',
  18.     `ts` bigint
  19. ) COMMENT '页面日志表'
  20. PARTITIONED BY (`dt` STRING)
  21. STORED AS PARQUET
  22. LOCATION '/warehouse/gmall/dwd/dwd_page_log'
  23. TBLPROPERTIES('parquet.compression'='lzo');

B、数据导入

  1. hive (gmall)>
  2. insert overwrite table dwd_page_log partition(dt='2020-06-14')
  3. select
  4.     get_json_object(line,'$.common.ar'),
  5.     get_json_object(line,'$.common.ba'),
  6.     get_json_object(line,'$.common.ch'),
  7.     get_json_object(line,'$.common.is_new'),
  8.     get_json_object(line,'$.common.md'),
  9.     get_json_object(line,'$.common.mid'),
  10.     get_json_object(line,'$.common.os'),
  11.     get_json_object(line,'$.common.uid'),
  12.     get_json_object(line,'$.common.vc'),
  13.     get_json_object(line,'$.page.during_time'),
  14.     get_json_object(line,'$.page.item'),
  15.     get_json_object(line,'$.page.item_type'),
  16.     get_json_object(line,'$.page.last_page_id'),
  17.     get_json_object(line,'$.page.page_id'),
  18.     get_json_object(line,'$.page.source_type'),
  19.     get_json_object(line,'$.ts')
  20. from ods_log
  21. where dt='2020-06-14'
  22. and get_json_object(line,'$.page') is not null;

C、查看数据

  1. hive (gmall)> 
  2. select * from dwd_page_log where dt='2020-06-14' limit 2;

5. 动作日志表

动作日志解析思路：动作日志表中每行数据对应用户的一个动作记录，一个动作记录应当包含公共信息、页面信息以及动作信息。先将包含action字段的日志过滤出来，然后通过 UDTF 函数，将action数组 “炸开”（类似于explode函数的效果），然后使用 get_json_object 函数解析每个字段。

动作日志表解析：

在这里插入图片描述

A、建表语句

  1. DROP TABLE IF EXISTS dwd_action_log;
  2. CREATE EXTERNAL TABLE dwd_action_log(
  3.     `area_code` STRING COMMENT '地区编码',
  4.     `brand` STRING COMMENT '手机品牌',
  5.     `channel` STRING COMMENT '渠道',
  6.     `is_new` STRING COMMENT '是否首次启动',
  7.     `model` STRING COMMENT '手机型号',
  8.     `mid_id` STRING COMMENT '设备id',
  9.     `os` STRING COMMENT '操作系统',
  10.     `user_id` STRING COMMENT '会员id',
  11.     `version_code` STRING COMMENT 'app版本号',
  12.     `during_time` BIGINT COMMENT '持续时间毫秒',
  13.     `page_item` STRING COMMENT '目标id ',
  14.     `page_item_type` STRING COMMENT '目标类型',
  15.     `last_page_id` STRING COMMENT '上页类型',
  16.     `page_id` STRING COMMENT '页面id ',
  17.     `source_type` STRING COMMENT '来源类型',
  18.     `action_id` STRING COMMENT '动作id',
  19.     `item` STRING COMMENT '目标id ',
  20.     `item_type` STRING COMMENT '目标类型',
  21.     `ts` BIGINT COMMENT '时间'
  22. ) COMMENT '动作日志表'
  23. PARTITIONED BY (`dt` STRING)
  24. STORED AS PARQUET
  25. LOCATION '/warehouse/gmall/dwd/dwd_action_log'
  26. TBLPROPERTIES('parquet.compression'='lzo');

B、创建 UDTF 函数 —— 设计思路

UDTF 思想：

在这里插入图片描述
UDTF 设计思路：

在这里插入图片描述

C、创建UDTF函数 —— 编写代码

（1）创建一个 maven 工程：hivefunction
（2）创建包名：com.fancy.hive.udtf
（3）引入如下依赖

  1. <dependencies>
  2.     <!--添加hive依赖-->
  3.     <dependency>
  4.         <groupId>org.apache.hive</groupId>
  5.         <artifactId>hive-exec</artifactId>
  6.         <version>3.1.2</version>
  7.     </dependency>
  8. </dependencies>

（4）编码

  1. package com.fancy.hive.udtf;
  3. import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
  4. import org.apache.hadoop.hive.ql.metadata.HiveException;
  5. import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
  6. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
  7. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
  8. import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
  9. import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
  10. import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
  11. import org.json.JSONArray;
  13. import java.util.ArrayList;
  14. import java.util.List;
  16. public class ExplodeJSONArray extends GenericUDTF {
  19.     @Override
  20.     public StructObjectInspector initialize(ObjectInspector[] argOIs) throws UDFArgumentException {
  23.         // 1 参数合法性检查
  24.         if (argOIs.length != 1) {
  26.             throw new UDFArgumentException("explode_json_array 只需要一个参数");
  27.         }
  29.         // 2 第一个参数必须为string
  30.         //判断参数是否为基础数据类型
  31.         if (argOIs[0].getCategory() != ObjectInspector.Category.PRIMITIVE) {
  33.             throw new UDFArgumentException("explode_json_array 只接受基础类型参数");
  34.         }
  36.         //将参数对象检查器强转为基础类型对象检查器
  37.         PrimitiveObjectInspector argumentOI = (PrimitiveObjectInspector) argOIs[0];
  39.         //判断参数是否为String类型
  40.         if (argumentOI.getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING) {
  42.             throw new UDFArgumentException("explode_json_array 只接受string类型的参数");
  43.         }
  45.         // 3 定义返回值名称和类型
  46.         List<String> fieldNames = new ArrayList<String>();
  47.         List<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
  49.         fieldNames.add("items");
  50.         fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
  52.         return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
  53.     }
  55.     public void process(Object[] objects) throws HiveException {
  58.         // 1 获取传入的数据
  59.         String jsonArray = objects[0].toString();
  61.         // 2 将string转换为json数组
  62.         JSONArray actions = new JSONArray(jsonArray);
  64.         // 3 循环一次，取出数组中的一个json，并写出
  65.         for (int i = 0; i < actions.length(); i++) {
  68.             String[] result = new String[1];
  69.             result[0] = actions.getString(i);
  70.             forward(result);
  71.         }
  72.     }
  74.     public void close() throws HiveException {
  77.     }
  79. }

D、创建函数

（1）打包

（2）将 hivefunction-1.0-SNAPSHOT.jar 上传到 node101 的 /opt/module，然后再将该 jar 包上传到 HDFS 的 /user/hive/jars 路径下

  1. [fancy@node101 module]$ hadoop fs -mkdir -p /user/hive/jars
  2. [fancy@node101 module]$ hadoop fs -put hivefunction-1.0-SNAPSHOT.jar /user/hive/jars

（3）创建永久函数与开发好的java class关联

  1. create function explode_json_array as 'com.fancy.hive.udtf.ExplodeJSONArray' using jar 'hdfs://hadoop102:8020/user/hive/jars/hivefunction-1.0-SNAPSHOT.jar';

注意：
如果修改了自定义函数重新生成 jar 包怎么处理？只需要替换HDFS 路径上的旧 jar 包，然后重启 Hive 客户端即可。

E、数据导入

  1. insert overwrite table dwd_action_log partition(dt='2020-06-14')
  2. select
  3.     get_json_object(line,'$.common.ar'),
  4.     get_json_object(line,'$.common.ba'),
  5.     get_json_object(line,'$.common.ch'),
  6.     get_json_object(line,'$.common.is_new'),
  7.     get_json_object(line,'$.common.md'),
  8.     get_json_object(line,'$.common.mid'),
  9.     get_json_object(line,'$.common.os'),
  10.     get_json_object(line,'$.common.uid'),
  11.     get_json_object(line,'$.common.vc'),
  12.     get_json_object(line,'$.page.during_time'),
  13.     get_json_object(line,'$.page.item'),
  14.     get_json_object(line,'$.page.item_type'),
  15.     get_json_object(line,'$.page.last_page_id'),
  16.     get_json_object(line,'$.page.page_id'),
  17.     get_json_object(line,'$.page.source_type'),
  18.     get_json_object(action,'$.action_id'),
  19.     get_json_object(action,'$.item'),
  20.     get_json_object(action,'$.item_type'),
  21.     get_json_object(action,'$.ts')
  22. from ods_log lateral view explode_json_array(get_json_object(line,'$.actions')) tmp as action
  23. where dt='2020-06-14'
  24. and get_json_object(line,'$.actions') is not null;

F、查看数据

  1. select * from dwd_action_log where dt='2020-06-14' limit 2;

6. 曝光日志表

曝光日志解析思路：曝光日志表中每行数据对应一个曝光记录，一个曝光记录应当包含公共信息、页面信息以及曝光信息。先将包含 display 字段的日志过滤出来，然后通过 UDTF 函数，将 display 数组 “炸开”（类似于 explode 函数的效果），然后使用 get_json_object 函数解析每个字段。

曝光日志表解析

在这里插入图片描述

A、建表语句

  1. DROP TABLE IF EXISTS dwd_display_log;
  2. CREATE EXTERNAL TABLE dwd_display_log(
  3.     `area_code` STRING COMMENT '地区编码',
  4.     `brand` STRING COMMENT '手机品牌',
  5.     `channel` STRING COMMENT '渠道',
  6.     `is_new` STRING COMMENT '是否首次启动',
  7.     `model` STRING COMMENT '手机型号',
  8.     `mid_id` STRING COMMENT '设备id',
  9.     `os` STRING COMMENT '操作系统',
  10.     `user_id` STRING COMMENT '会员id',
  11.     `version_code` STRING COMMENT 'app版本号',
  12.     `during_time` BIGINT COMMENT 'app版本号',
  13.     `page_item` STRING COMMENT '目标id ',
  14.     `page_item_type` STRING COMMENT '目标类型',
  15.     `last_page_id` STRING COMMENT '上页类型',
  16.     `page_id` STRING COMMENT '页面ID ',
  17.     `source_type` STRING COMMENT '来源类型',
  18.     `ts` BIGINT COMMENT 'app版本号',
  19.     `display_type` STRING COMMENT '曝光类型',
  20.     `item` STRING COMMENT '曝光对象id ',
  21.     `item_type` STRING COMMENT 'app版本号',
  22.     `order` BIGINT COMMENT '曝光顺序',
  23.     `pos_id` BIGINT COMMENT '曝光位置'
  24. ) COMMENT '曝光日志表'
  25. PARTITIONED BY (`dt` STRING)
  26. STORED AS PARQUET
  27. LOCATION '/warehouse/gmall/dwd/dwd_display_log'
  28. TBLPROPERTIES('parquet.compression'='lzo');

B、数据导入

  1. insert overwrite table dwd_display_log partition(dt='2020-06-14')
  2. select
  3.     get_json_object(line,'$.common.ar'),
  4.     get_json_object(line,'$.common.ba'),
  5.     get_json_object(line,'$.common.ch'),
  6.     get_json_object(line,'$.common.is_new'),
  7.     get_json_object(line,'$.common.md'),
  8.     get_json_object(line,'$.common.mid'),
  9.     get_json_object(line,'$.common.os'),
  10.     get_json_object(line,'$.common.uid'),
  11.     get_json_object(line,'$.common.vc'),
  12.     get_json_object(line,'$.page.during_time'),
  13.     get_json_object(line,'$.page.item'),
  14.     get_json_object(line,'$.page.item_type'),
  15.     get_json_object(line,'$.page.last_page_id'),
  16.     get_json_object(line,'$.page.page_id'),
  17.     get_json_object(line,'$.page.source_type'),
  18.     get_json_object(line,'$.ts'),
  19.     get_json_object(display,'$.display_type'),
  20.     get_json_object(display,'$.item'),
  21.     get_json_object(display,'$.item_type'),
  22.     get_json_object(display,'$.order'),
  23.     get_json_object(display,'$.pos_id')
  24. from ods_log lateral view explode_json_array(get_json_object(line,'$.displays')) tmp as display
  25. where dt='2020-06-14'
  26. and get_json_object(line,'$.displays') is not null;

C、查看数据

  1. select * from dwd_display_log where dt='2020-06-14' limit 2;

7. 错误日志表

错误日志解析思路：错误日志表中每行数据对应一个错误记录，为方便定位错误，一个错误记录应当包含与之对应的公共信息、页面信息、曝光信息、动作信息、启动信息以及错误信息。先将包含err字段的日志过滤出来，然后使用 get_json_object 函数解析所有字段。

在这里插入图片描述

A、建表语句

  1. DROP TABLE IF EXISTS dwd_error_log;
  2. CREATE EXTERNAL TABLE dwd_error_log(
  3.     `area_code` STRING COMMENT '地区编码',
  4.     `brand` STRING COMMENT '手机品牌',
  5.     `channel` STRING COMMENT '渠道',
  6.     `is_new` STRING COMMENT '是否首次启动',
  7.     `model` STRING COMMENT '手机型号',
  8.     `mid_id` STRING COMMENT '设备id',
  9.     `os` STRING COMMENT '操作系统',
  10.     `user_id` STRING COMMENT '会员id',
  11.     `version_code` STRING COMMENT 'app版本号',
  12.     `page_item` STRING COMMENT '目标id ',
  13.     `page_item_type` STRING COMMENT '目标类型',
  14.     `last_page_id` STRING COMMENT '上页类型',
  15.     `page_id` STRING COMMENT '页面ID ',
  16.     `source_type` STRING COMMENT '来源类型',
  17.     `entry` STRING COMMENT ' icon手机图标  notice 通知 install 安装后启动',
  18.     `loading_time` STRING COMMENT '启动加载时间',
  19.     `open_ad_id` STRING COMMENT '广告页ID ',
  20.     `open_ad_ms` STRING COMMENT '广告总共播放时间',
  21.     `open_ad_skip_ms` STRING COMMENT '用户跳过广告时点',
  22.     `actions` STRING COMMENT '动作',
  23.     `displays` STRING COMMENT '曝光',
  24.     `ts` STRING COMMENT '时间',
  25.     `error_code` STRING COMMENT '错误码',
  26.     `msg` STRING COMMENT '错误信息'
  27. ) COMMENT '错误日志表'
  28. PARTITIONED BY (`dt` STRING)
  29. STORED AS PARQUET
  30. LOCATION '/warehouse/gmall/dwd/dwd_error_log'
  31. TBLPROPERTIES('parquet.compression'='lzo');

说明：此处为对动作数组和曝光数组做处理，如需分析错误与单个动作或曝光的关联，可先使用 explode_json_array 函数将数组“炸开”，再使用get_json_object 函数获取具体字段。

D、数据导入

  1. insert overwrite table dwd_error_log partition(dt='2020-06-14')
  2. select
  3.     get_json_object(line,'$.common.ar'),
  4.     get_json_object(line,'$.common.ba'),
  5.     get_json_object(line,'$.common.ch'),
  6.     get_json_object(line,'$.common.is_new'),
  7.     get_json_object(line,'$.common.md'),
  8.     get_json_object(line,'$.common.mid'),
  9.     get_json_object(line,'$.common.os'),
  10.     get_json_object(line,'$.common.uid'),
  11.     get_json_object(line,'$.common.vc'),
  12.     get_json_object(line,'$.page.item'),
  13.     get_json_object(line,'$.page.item_type'),
  14.     get_json_object(line,'$.page.last_page_id'),
  15.     get_json_object(line,'$.page.page_id'),
  16.     get_json_object(line,'$.page.source_type'),
  17.     get_json_object(line,'$.start.entry'),
  18.     get_json_object(line,'$.start.loading_time'),
  19.     get_json_object(line,'$.start.open_ad_id'),
  20.     get_json_object(line,'$.start.open_ad_ms'),
  21.     get_json_object(line,'$.start.open_ad_skip_ms'),
  22.     get_json_object(line,'$.actions'),
  23.     get_json_object(line,'$.displays'),
  24.     get_json_object(line,'$.ts'),
  25.     get_json_object(line,'$.err.error_code'),
  26.     get_json_object(line,'$.err.msg')
  27. from ods_log
  28. where dt='2020-06-14'
  29. and get_json_object(line,'$.err') is not null;

D、查看数据

  1. hive (gmall)> 
  2. select * from dwd_error_log where dt='2020-06-14' limit 2;

8. DWD 层用户行为数据加载脚本

A、编写脚本

在 node101 的 /home/fancy/bin 目录下创建脚本

  1. [fancy@node101 bin]$ vim ods_to_dwd_log.sh

在脚本中编写如下内容

  1. #!/bin/bash
  3. APP=gmall
  4. # 如果是输入的日期按照取输入日期；如果没输入日期取当前时间的前一天
  5. if [ -n "$2" ] ;then
  6.     do_date=$2
  7. else 
  8.     do_date=`date -d "-1 day" +%F`
  9. fi
  11. dwd_start_log="
  12. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  13. insert overwrite table ${APP}.dwd_start_log partition(dt='$do_date')
  14. select
  15.     get_json_object(line,'$.common.ar'),
  16.     get_json_object(line,'$.common.ba'),
  17.     get_json_object(line,'$.common.ch'),
  18.     get_json_object(line,'$.common.is_new'),
  19.     get_json_object(line,'$.common.md'),
  20.     get_json_object(line,'$.common.mid'),
  21.     get_json_object(line,'$.common.os'),
  22.     get_json_object(line,'$.common.uid'),
  23.     get_json_object(line,'$.common.vc'),
  24.     get_json_object(line,'$.start.entry'),
  25.     get_json_object(line,'$.start.loading_time'),
  26.     get_json_object(line,'$.start.open_ad_id'),
  27.     get_json_object(line,'$.start.open_ad_ms'),
  28.     get_json_object(line,'$.start.open_ad_skip_ms'),
  29.     get_json_object(line,'$.ts')
  30. from ${APP}.ods_log
  31. where dt='$do_date'
  32. and get_json_object(line,'$.start') is not null;"
  34. dwd_page_log="
  35. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  36. insert overwrite table ${APP}.dwd_page_log partition(dt='$do_date')
  37. select
  38.     get_json_object(line,'$.common.ar'),
  39.     get_json_object(line,'$.common.ba'),
  40.     get_json_object(line,'$.common.ch'),
  41.     get_json_object(line,'$.common.is_new'),
  42.     get_json_object(line,'$.common.md'),
  43.     get_json_object(line,'$.common.mid'),
  44.     get_json_object(line,'$.common.os'),
  45.     get_json_object(line,'$.common.uid'),
  46.     get_json_object(line,'$.common.vc'),
  47.     get_json_object(line,'$.page.during_time'),
  48.     get_json_object(line,'$.page.item'),
  49.     get_json_object(line,'$.page.item_type'),
  50.     get_json_object(line,'$.page.last_page_id'),
  51.     get_json_object(line,'$.page.page_id'),
  52.     get_json_object(line,'$.page.source_type'),
  53.     get_json_object(line,'$.ts')
  54. from ${APP}.ods_log
  55. where dt='$do_date'
  56. and get_json_object(line,'$.page') is not null;"
  58. dwd_action_log="
  59. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  60. insert overwrite table ${APP}.dwd_action_log partition(dt='$do_date')
  61. select
  62.     get_json_object(line,'$.common.ar'),
  63.     get_json_object(line,'$.common.ba'),
  64.     get_json_object(line,'$.common.ch'),
  65.     get_json_object(line,'$.common.is_new'),
  66.     get_json_object(line,'$.common.md'),
  67.     get_json_object(line,'$.common.mid'),
  68.     get_json_object(line,'$.common.os'),
  69.     get_json_object(line,'$.common.uid'),
  70.     get_json_object(line,'$.common.vc'),
  71.     get_json_object(line,'$.page.during_time'),
  72.     get_json_object(line,'$.page.item'),
  73.     get_json_object(line,'$.page.item_type'),
  74.     get_json_object(line,'$.page.last_page_id'),
  75.     get_json_object(line,'$.page.page_id'),
  76.     get_json_object(line,'$.page.source_type'),
  77.     get_json_object(action,'$.action_id'),
  78.     get_json_object(action,'$.item'),
  79.     get_json_object(action,'$.item_type'),
  80.     get_json_object(action,'$.ts')
  81. from ${APP}.ods_log lateral view ${APP}.explode_json_array(get_json_object(line,'$.actions')) tmp as action
  82. where dt='$do_date'
  83. and get_json_object(line,'$.actions') is not null;"
  86. dwd_display_log="
  87. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  88. insert overwrite table ${APP}.dwd_display_log partition(dt='$do_date')
  89. select
  90.     get_json_object(line,'$.common.ar'),
  91.     get_json_object(line,'$.common.ba'),
  92.     get_json_object(line,'$.common.ch'),
  93.     get_json_object(line,'$.common.is_new'),
  94.     get_json_object(line,'$.common.md'),
  95.     get_json_object(line,'$.common.mid'),
  96.     get_json_object(line,'$.common.os'),
  97.     get_json_object(line,'$.common.uid'),
  98.     get_json_object(line,'$.common.vc'),
  99.     get_json_object(line,'$.page.during_time'),
  100.     get_json_object(line,'$.page.item'),
  101.     get_json_object(line,'$.page.item_type'),
  102.     get_json_object(line,'$.page.last_page_id'),
  103.     get_json_object(line,'$.page.page_id'),
  104.     get_json_object(line,'$.page.source_type'),
  105.     get_json_object(line,'$.ts'),
  106.     get_json_object(display,'$.display_type'),
  107.     get_json_object(display,'$.item'),
  108.     get_json_object(display,'$.item_type'),
  109.     get_json_object(display,'$.order'),
  110.     get_json_object(display,'$.pos_id')
  111. from ${APP}.ods_log lateral view ${APP}.explode_json_array(get_json_object(line,'$.displays')) tmp as display
  112. where dt='$do_date'
  113. and get_json_object(line,'$.displays') is not null;"
  116. dwd_error_log="
  117. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  118. insert overwrite table ${APP}.dwd_error_log partition(dt='$do_date')
  119. select
  120.     get_json_object(line,'$.common.ar'),
  121.     get_json_object(line,'$.common.ba'),
  122.     get_json_object(line,'$.common.ch'),
  123.     get_json_object(line,'$.common.is_new'),
  124.     get_json_object(line,'$.common.md'),
  125.     get_json_object(line,'$.common.mid'),
  126.     get_json_object(line,'$.common.os'),
  127.     get_json_object(line,'$.common.uid'),
  128.     get_json_object(line,'$.common.vc'),
  129.     get_json_object(line,'$.page.item'),
  130.     get_json_object(line,'$.page.item_type'),
  131.     get_json_object(line,'$.page.last_page_id'),
  132.     get_json_object(line,'$.page.page_id'),
  133.     get_json_object(line,'$.page.source_type'),
  134.     get_json_object(line,'$.start.entry'),
  135.     get_json_object(line,'$.start.loading_time'),
  136.     get_json_object(line,'$.start.open_ad_id'),
  137.     get_json_object(line,'$.start.open_ad_ms'),
  138.     get_json_object(line,'$.start.open_ad_skip_ms'),
  139.     get_json_object(line,'$.actions'),
  140.     get_json_object(line,'$.displays'),
  141.     get_json_object(line,'$.ts'),
  142.     get_json_object(line,'$.err.error_code'),
  143.     get_json_object(line,'$.err.msg')
  144. from ${APP}.ods_log
  145. where dt='$do_date'
  146. and get_json_object(line,'$.err') is not null;"
  149. case $1 in
  150.     dwd_start_log )
  151.         hive -e "$dwd_start_log"
  152.     ;;
  153.     dwd_page_log )
  154.         hive -e "$dwd_page_log"
  155.     ;;
  156.     dwd_action_log )
  157.         hive -e "$dwd_action_log"
  158.     ;;
  159.     dwd_display_log )
  160.         hive -e "$dwd_display_log"
  161.     ;;
  162.     dwd_error_log )
  163.         hive -e "$dwd_error_log"
  164.     ;;
  165.     all )
  166.         hive -e "$dwd_start_log$dwd_page_log$dwd_action_log$dwd_display_log$dwd_error_log"
  167.     ;;
  168. esac

增加脚本执行权限

  1. [fancy@node101 bin]$ chmod 777 ods_to_dwd_log.sh

B、脚本使用

执行脚本

  1. [fancy@node101 module]$ ods_to_dwd_log.sh all 2020-06-14

查询导入结果

二、DWD层 (业务数据)

业务数据方面DWD层的搭建主要注意点在于维度建模。

1. 评价事实表 (事务型事实表)

A、建表语句

  1. DROP TABLE IF EXISTS dwd_comment_info;
  2. CREATE EXTERNAL TABLE dwd_comment_info(
  3.     `id` STRING COMMENT '编号',
  4.     `user_id` STRING COMMENT '用户ID',
  5.     `sku_id` STRING COMMENT '商品sku',
  6.     `spu_id` STRING COMMENT '商品spu',
  7.     `order_id` STRING COMMENT '订单ID',
  8.     `appraise` STRING COMMENT '评价(好评、中评、差评、默认评价)',
  9.     `create_time` STRING COMMENT '评价时间'
  10. ) COMMENT '评价事实表'
  11. PARTITIONED BY (`dt` STRING)
  12. STORED AS PARQUET
  13. LOCATION '/warehouse/gmall/dwd/dwd_comment_info/'
  14. TBLPROPERTIES ("parquet.compression"="lzo");

B、分区规划

在这里插入图片描述

C、数据装载

在这里插入图片描述
1.首日装载

  1. insert overwrite table dwd_comment_info partition (dt)
  2. select
  3.     id,
  4.     user_id,
  5.     sku_id,
  6.     spu_id,
  7.     order_id,
  8.     appraise,
  9.     create_time,
  10.     date_format(create_time,'yyyy-MM-dd')
  11. from ods_comment_info
  12. where dt='2020-06-14';

2.每日装载

  1. insert overwrite table dwd_comment_info partition(dt='2020-06-15')
  2. select
  3.     id,
  4.     user_id,
  5.     sku_id,
  6.     spu_id,
  7.     order_id,
  8.     appraise,
  9.     create_time
  10. from ods_comment_info where dt='2020-06-15';

2. 订单明细事实表 (事务型事实表)

A、建表语句

  1. DROP TABLE IF EXISTS dwd_order_detail;
  2. CREATE EXTERNAL TABLE dwd_order_detail (
  3.     `id` STRING COMMENT '订单编号',
  4.     `order_id` STRING COMMENT '订单号',
  5.     `user_id` STRING COMMENT '用户id',
  6.     `sku_id` STRING COMMENT 'sku商品id',
  7.     `province_id` STRING COMMENT '省份ID',
  8.     `activity_id` STRING COMMENT '活动ID',
  9.     `activity_rule_id` STRING COMMENT '活动规则ID',
  10.     `coupon_id` STRING COMMENT '优惠券ID',
  11.     `create_time` STRING COMMENT '创建时间',
  12.     `source_type` STRING COMMENT '来源类型',
  13.     `source_id` STRING COMMENT '来源编号',
  14.     `sku_num` BIGINT COMMENT '商品数量',
  15.     `original_amount` DECIMAL(16,2) COMMENT '原始价格',
  16.     `split_activity_amount` DECIMAL(16,2) COMMENT '活动优惠分摊',
  17.     `split_coupon_amount` DECIMAL(16,2) COMMENT '优惠券优惠分摊',
  18.     `split_final_amount` DECIMAL(16,2) COMMENT '最终价格分摊'
  19. ) COMMENT '订单明细事实表表'
  20. PARTITIONED BY (`dt` STRING)
  21. STORED AS PARQUET
  22. LOCATION '/warehouse/gmall/dwd/dwd_order_detail/'
  23. TBLPROPERTIES ("parquet.compression"="lzo");

B、分区规划

在这里插入图片描述

C、数据装载

在这里插入图片描述

1.首日装载

  1. insert overwrite table dwd_order_detail partition(dt)
  2. select
  3.     od.id,
  4.     od.order_id,
  5.     oi.user_id,
  6.     od.sku_id,
  7.     oi.province_id,
  8.     oda.activity_id,
  9.     oda.activity_rule_id,
  10.     odc.coupon_id,
  11.     od.create_time,
  12.     od.source_type,
  13.     od.source_id,
  14.     od.sku_num,
  15.     od.order_price*od.sku_num,
  16.     od.split_activity_amount,
  17.     od.split_coupon_amount,
  18.     od.split_final_amount,
  19.     date_format(create_time,'yyyy-MM-dd')
  20. from
  21. (
  22.     select
  23.         *
  24.     from ods_order_detail
  25.     where dt='2020-06-14'
  26. )od
  27. left join
  28. (
  29.     select
  30.         id,
  31.         user_id,
  32.         province_id
  33.     from ods_order_info
  34.     where dt='2020-06-14'
  35. )oi
  36. on od.order_id=oi.id
  37. left join
  38. (
  39.     select
  40.         order_detail_id,
  41.         activity_id,
  42.         activity_rule_id
  43.     from ods_order_detail_activity
  44.     where dt='2020-06-14'
  45. )oda
  46. on od.id=oda.order_detail_id
  47. left join
  48. (
  49.     select
  50.         order_detail_id,
  51.         coupon_id
  52.     from ods_order_detail_coupon
  53.     where dt='2020-06-14'
  54. )odc
  55. on od.id=odc.order_detail_id;

2.每日装载

  1. insert overwrite table dwd_order_detail partition(dt='2020-06-15')
  2. select
  3.     od.id,
  4.     od.order_id,
  5.     oi.user_id,
  6.     od.sku_id,
  7.     oi.province_id,
  8.     oda.activity_id,
  9.     oda.activity_rule_id,
  10.     odc.coupon_id,
  11.     od.create_time,
  12.     od.source_type,
  13.     od.source_id,
  14.     od.sku_num,
  15.     od.order_price*od.sku_num,
  16.     od.split_activity_amount,
  17.     od.split_coupon_amount,
  18.     od.split_final_amount
  19. from
  20. (
  21.     select
  22.         *
  23.     from ods_order_detail
  24.     where dt='2020-06-15'
  25. )od
  26. left join
  27. (
  28.     select
  29.         id,
  30.         user_id,
  31.         province_id
  32.     from ods_order_info
  33.     where dt='2020-06-15'
  34. )oi
  35. on od.order_id=oi.id
  36. left join
  37. (
  38.     select
  39.         order_detail_id,
  40.         activity_id,
  41.         activity_rule_id
  42.     from ods_order_detail_activity
  43.     where dt='2020-06-15'
  44. )oda
  45. on od.id=oda.order_detail_id
  46. left join
  47. (
  48.     select
  49.         order_detail_id,
  50.         coupon_id
  51.     from ods_order_detail_coupon
  52.     where dt='2020-06-15'
  53. )odc
  54. on od.id=odc.order_detail_id;

3. 退单事实表 (事务型事实表)

A、建表语句

  1. DROP TABLE IF EXISTS dwd_order_refund_info;
  2. CREATE EXTERNAL TABLE dwd_order_refund_info(
  3.     `id` STRING COMMENT '编号',
  4.     `user_id` STRING COMMENT '用户ID',
  5.     `order_id` STRING COMMENT '订单ID',
  6.     `sku_id` STRING COMMENT '商品ID',
  7.     `province_id` STRING COMMENT '地区ID',
  8.     `refund_type` STRING COMMENT '退单类型',
  9.     `refund_num` BIGINT COMMENT '退单件数',
  10.     `refund_amount` DECIMAL(16,2) COMMENT '退单金额',
  11.     `refund_reason_type` STRING COMMENT '退单原因类型',
  12.     `create_time` STRING COMMENT '退单时间'
  13. ) COMMENT '退单事实表'
  14. PARTITIONED BY (`dt` STRING)
  15. STORED AS PARQUET
  16. LOCATION '/warehouse/gmall/dwd/dwd_order_refund_info/'
  17. TBLPROPERTIES ("parquet.compression"="lzo");

B、分区规划
在这里插入图片描述

C、数据装载

在这里插入图片描述

1.首日装载

  1. insert overwrite table dwd_order_refund_info partition(dt)
  2. select
  3.     ri.id,
  4.     ri.user_id,
  5.     ri.order_id,
  6.     ri.sku_id,
  7.     oi.province_id,
  8.     ri.refund_type,
  9.     ri.refund_num,
  10.     ri.refund_amount,
  11.     ri.refund_reason_type,
  12.     ri.create_time,
  13.     date_format(ri.create_time,'yyyy-MM-dd')
  14. from
  15. (
  16.     select * from ods_order_refund_info where dt='2020-06-14'
  17. )ri
  18. left join
  19. (
  20.     select id,province_id from ods_order_info where dt='2020-06-14'
  21. )oi
  22. on ri.order_id=oi.id;

2.每日装载

  1. insert overwrite table dwd_order_refund_info partition(dt='2020-06-15')
  2. select
  3.     ri.id,
  4.     ri.user_id,
  5.     ri.order_id,
  6.     ri.sku_id,
  7.     oi.province_id,
  8.     ri.refund_type,
  9.     ri.refund_num,
  10.     ri.refund_amount,
  11.     ri.refund_reason_type,
  12.     ri.create_time
  13. from
  14. (
  15.     select * from ods_order_refund_info where dt='2020-06-15'
  16. )ri
  17. left join
  18. (
  19.     select id,province_id from ods_order_info where dt='2020-06-15'
  20. )oi
  21. on ri.order_id=oi.id;

3.查询加载结果

4. 加购事实表 (周期型快照事实表，每日快照)

A、建表语句

  1. DROP TABLE IF EXISTS dwd_cart_info;
  2. CREATE EXTERNAL TABLE dwd_cart_info(
  3.     `id` STRING COMMENT '编号',
  4.     `user_id` STRING COMMENT '用户ID',
  5.     `sku_id` STRING COMMENT '商品ID',
  6.     `source_type` STRING COMMENT '来源类型',
  7.     `source_id` STRING COMMENT '来源编号',
  8.     `cart_price` DECIMAL(16,2) COMMENT '加入购物车时的价格',
  9.     `is_ordered` STRING COMMENT '是否已下单',
  10.     `create_time` STRING COMMENT '创建时间',
  11.     `operate_time` STRING COMMENT '修改时间',
  12.     `order_time` STRING COMMENT '下单时间',
  13.     `sku_num` BIGINT COMMENT '加购数量'
  14. ) COMMENT '加购事实表'
  15. PARTITIONED BY (`dt` STRING)
  16. STORED AS PARQUET
  17. LOCATION '/warehouse/gmall/dwd/dwd_cart_info/'
  18. TBLPROPERTIES ("parquet.compression"="lzo");

B、分区规划

在这里插入图片描述

C、数据装载

在这里插入图片描述

1.首日装载

  1. insert overwrite table dwd_cart_info partition(dt='2020-06-14')
  2. select
  3.     id,
  4.     user_id,
  5.     sku_id,
  6.     source_type,
  7.     source_id,
  8.     cart_price,
  9.     is_ordered,
  10.     create_time,
  11.     operate_time,
  12.     order_time,
  13.     sku_num
  14. from ods_cart_info
  15. where dt='2020-06-14';

2.每日装载

  1. insert overwrite table dwd_cart_info partition(dt='2020-06-15')
  2. select
  3.     id,
  4.     user_id,
  5.     sku_id,
  6.     source_type,
  7.     source_id,
  8.     cart_price,
  9.     is_ordered,
  10.     create_time,
  11.     operate_time,
  12.     order_time,
  13.     sku_num
  14. from ods_cart_info
  15. where dt='2020-06-15';

5. 收藏事实表 (周期型快照事实表，每日快照)

A、建表语句

  1. DROP TABLE IF EXISTS dwd_favor_info;
  2. CREATE EXTERNAL TABLE dwd_favor_info(
  3.     `id` STRING COMMENT '编号',
  4.     `user_id` STRING  COMMENT '用户id',
  5.     `sku_id` STRING  COMMENT 'skuid',
  6.     `spu_id` STRING  COMMENT 'spuid',
  7.     `is_cancel` STRING  COMMENT '是否取消',
  8.     `create_time` STRING  COMMENT '收藏时间',
  9.     `cancel_time` STRING  COMMENT '取消时间'
  10. ) COMMENT '收藏事实表'
  11. PARTITIONED BY (`dt` STRING)
  12. STORED AS PARQUET
  13. LOCATION '/warehouse/gmall/dwd/dwd_favor_info/'
  14. TBLPROPERTIES ("parquet.compression"="lzo");

B、分区规划

在这里插入图片描述

C、数据装载

在这里插入图片描述

1.首日装载

  1. insert overwrite table dwd_favor_info partition(dt='2020-06-14')
  2. select
  3.     id,
  4.     user_id,
  5.     sku_id,
  6.     spu_id,
  7.     is_cancel,
  8.     create_time,
  9.     cancel_time
  10. from ods_favor_info
  11. where dt='2020-06-14';

2.每日装载

  1. insert overwrite table dwd_favor_info partition(dt='2020-06-15')
  2. select
  3.     id,
  4.     user_id,
  5.     sku_id,
  6.     spu_id,
  7.     is_cancel,
  8.     create_time,
  9.     cancel_time
  10. from ods_favor_info
  11. where dt='2020-06-15';

6. 优惠券领用事实表 (累积型快照事实表)

A、建表语句

  1. DROP TABLE IF EXISTS dwd_coupon_use;
  2. CREATE EXTERNAL TABLE dwd_coupon_use(
  3.     `id` STRING COMMENT '编号',
  4.     `coupon_id` STRING  COMMENT '优惠券ID',
  5.     `user_id` STRING  COMMENT 'userid',
  6.     `order_id` STRING  COMMENT '订单id',
  7.     `coupon_status` STRING  COMMENT '优惠券状态',
  8.     `get_time` STRING  COMMENT '领取时间',
  9.     `using_time` STRING  COMMENT '使用时间(下单)',
  10.     `used_time` STRING  COMMENT '使用时间(支付)',
  11.     `expire_time` STRING COMMENT '过期时间'
  12. ) COMMENT '优惠券领用事实表'
  13. PARTITIONED BY (`dt` STRING)
  14. STORED AS PARQUET
  15. LOCATION '/warehouse/gmall/dwd/dwd_coupon_use/'
  16. TBLPROPERTIES ("parquet.compression"="lzo");

B、分区规划
在这里插入图片描述

C、数据装载

在这里插入图片描述

1.首日装载

  1. insert overwrite table dwd_coupon_use partition(dt)
  2. select
  3.     id,
  4.     coupon_id,
  5.     user_id,
  6.     order_id,
  7.     coupon_status,
  8.     get_time,
  9.     using_time,
  10.     used_time,
  11.     expire_time,
  12.     coalesce(date_format(used_time,'yyyy-MM-dd'),date_format(expire_time,'yyyy-MM-dd'),'9999-99-99')
  13. from ods_coupon_use
  14. where dt='2020-06-14';

2.每日装载

a.装载逻辑
在这里插入图片描述
b.转载语句

  1. insert overwrite table dwd_coupon_use partition(dt)
  2. select
  3.     nvl(new.id,old.id),
  4.     nvl(new.coupon_id,old.coupon_id),
  5.     nvl(new.user_id,old.user_id),
  6.     nvl(new.order_id,old.order_id),
  7.     nvl(new.coupon_status,old.coupon_status),
  8.     nvl(new.get_time,old.get_time),
  9.     nvl(new.using_time,old.using_time),
  10.     nvl(new.used_time,old.used_time),
  11.     nvl(new.expire_time,old.expire_time),
  12.     coalesce(date_format(nvl(new.used_time,old.used_time),'yyyy-MM-dd'),date_format(nvl(new.expire_time,old.expire_time),'yyyy-MM-dd'),'9999-99-99')
  13. from
  14. (
  15.     select
  16.         id,
  17.         coupon_id,
  18.         user_id,
  19.         order_id,
  20.         coupon_status,
  21.         get_time,
  22.         using_time,
  23.         used_time,
  24.         expire_time
  25.     from dwd_coupon_use
  26.     where dt='9999-99-99'
  27. )old
  28. full outer join
  29. (
  30.     select
  31.         id,
  32.         coupon_id,
  33.         user_id,
  34.         order_id,
  35.         coupon_status,
  36.         get_time,
  37.         using_time,
  38.         used_time,
  39.         expire_time
  40.     from ods_coupon_use
  41.     where dt='2020-06-15'
  42. )new
  43. on old.id=new.id;

7. 支付事实表 (累积型快照事实表)

A、建表语句

  1. DROP TABLE IF EXISTS dwd_payment_info;
  2. CREATE EXTERNAL TABLE dwd_payment_info (
  3.     `id` STRING COMMENT '编号',
  4.     `order_id` STRING COMMENT '订单编号',
  5.     `user_id` STRING COMMENT '用户编号',
  6.     `province_id` STRING COMMENT '地区ID',
  7.     `trade_no` STRING COMMENT '交易编号',
  8.     `out_trade_no` STRING COMMENT '对外交易编号',
  9.     `payment_type` STRING COMMENT '支付类型',
  10.     `payment_amount` DECIMAL(16,2) COMMENT '支付金额',
  11.     `payment_status` STRING COMMENT '支付状态',
  12.     `create_time` STRING COMMENT '创建时间',--调用第三方支付接口的时间
  13.     `callback_time` STRING COMMENT '完成时间'--支付完成时间，即支付成功回调时间
  14. ) COMMENT '支付事实表表'
  15. PARTITIONED BY (`dt` STRING)
  16. STORED AS PARQUET
  17. LOCATION '/warehouse/gmall/dwd/dwd_payment_info/'
  18. TBLPROPERTIES ("parquet.compression"="lzo");

B、分区规划
在这里插入图片描述

C、数据装载

在这里插入图片描述

1.首日装载

  1. insert overwrite table dwd_payment_info partition(dt)
  2. select
  3.     pi.id,
  4.     pi.order_id,
  5.     pi.user_id,
  6.     oi.province_id,
  7.     pi.trade_no,
  8.     pi.out_trade_no,
  9.     pi.payment_type,
  10.     pi.payment_amount,
  11.     pi.payment_status,
  12.     pi.create_time,
  13.     pi.callback_time,
  14.     nvl(date_format(pi.callback_time,'yyyy-MM-dd'),'9999-99-99')
  15. from
  16. (
  17.     select * from ods_payment_info where dt='2020-06-14'
  18. )pi
  19. left join
  20. (
  21.     select id,province_id from ods_order_info where dt='2020-06-14'
  22. )oi
  23. on pi.order_id=oi.id;

2.每日装载

  1. insert overwrite table dwd_payment_info partition(dt)
  2. select
  3.     nvl(new.id,old.id),
  4.     nvl(new.order_id,old.order_id),
  5.     nvl(new.user_id,old.user_id),
  6.     nvl(new.province_id,old.province_id),
  7.     nvl(new.trade_no,old.trade_no),
  8.     nvl(new.out_trade_no,old.out_trade_no),
  9.     nvl(new.payment_type,old.payment_type),
  10.     nvl(new.payment_amount,old.payment_amount),
  11.     nvl(new.payment_status,old.payment_status),
  12.     nvl(new.create_time,old.create_time),
  13.     nvl(new.callback_time,old.callback_time),
  14.     nvl(date_format(nvl(new.callback_time,old.callback_time),'yyyy-MM-dd'),'9999-99-99')
  15. from
  16. (
  17.     select id,
  18.        order_id,
  19.        user_id,
  20.        province_id,
  21.        trade_no,
  22.        out_trade_no,
  23.        payment_type,
  24.        payment_amount,
  25.        payment_status,
  26.        create_time,
  27.        callback_time
  28.     from dwd_payment_info
  29.     where dt = '9999-99-99'
  30. )old
  31. full outer join
  32. (
  33.     select
  34.         pi.id,
  35.         pi.out_trade_no,
  36.         pi.order_id,
  37.         pi.user_id,
  38.         oi.province_id,
  39.         pi.payment_type,
  40.         pi.trade_no,
  41.         pi.payment_amount,
  42.         pi.payment_status,
  43.         pi.create_time,
  44.         pi.callback_time
  45.     from
  46.     (
  47.         select * from ods_payment_info where dt='2020-06-15'
  48.     )pi
  49.     left join
  50.     (
  51.         select id,province_id from ods_order_info where dt='2020-06-15'
  52.     )oi
  53.     on pi.order_id=oi.id
  54. )new
  55. on old.id=new.id;

8. 退款事实表 (累积型快照事实表)

A、建表语句

  1. DROP TABLE IF EXISTS dwd_refund_payment;
  2. CREATE EXTERNAL TABLE dwd_refund_payment (
  3.     `id` STRING COMMENT '编号',
  4.     `user_id` STRING COMMENT '用户ID',
  5.     `order_id` STRING COMMENT '订单编号',
  6.     `sku_id` STRING COMMENT 'SKU编号',
  7.     `province_id` STRING COMMENT '地区ID',
  8.     `trade_no` STRING COMMENT '交易编号',
  9.     `out_trade_no` STRING COMMENT '对外交易编号',
  10.     `payment_type` STRING COMMENT '支付类型',
  11.     `refund_amount` DECIMAL(16,2) COMMENT '退款金额',
  12.     `refund_status` STRING COMMENT '退款状态',
  13.     `create_time` STRING COMMENT '创建时间',--调用第三方支付接口的时间
  14.     `callback_time` STRING COMMENT '回调时间'--支付接口回调时间，即支付成功时间
  15. ) COMMENT '退款事实表'
  16. PARTITIONED BY (`dt` STRING)
  17. STORED AS PARQUET
  18. LOCATION '/warehouse/gmall/dwd/dwd_refund_payment/'
  19. TBLPROPERTIES ("parquet.compression"="lzo");

B、分区规划
在这里插入图片描述

C、数据装载

在这里插入图片描述

1.首日装载

  1. insert overwrite table dwd_refund_payment partition(dt)
  2. select
  3.     rp.id,
  4.     user_id,
  5.     order_id,
  6.     sku_id,
  7.     province_id,
  8.     trade_no,
  9.     out_trade_no,
  10.     payment_type,
  11.     refund_amount,
  12.     refund_status,
  13.     create_time,
  14.     callback_time,
  15.     nvl(date_format(callback_time,'yyyy-MM-dd'),'9999-99-99')
  16. from
  17. (
  18.     select
  19.         id,
  20.         out_trade_no,
  21.         order_id,
  22.         sku_id,
  23.         payment_type,
  24.         trade_no,
  25.         refund_amount,
  26.         refund_status,
  27.         create_time,
  28.         callback_time
  29.     from ods_refund_payment
  30.     where dt='2020-06-14'
  31. )rp
  32. left join
  33. (
  34.     select
  35.         id,
  36.         user_id,
  37.         province_id
  38.     from ods_order_info
  39.     where dt='2020-06-14'
  40. )oi
  41. on rp.order_id=oi.id;

2.每日装载

  1. insert overwrite table dwd_refund_payment partition(dt)
  2. select
  3.     nvl(new.id,old.id),
  4.     nvl(new.user_id,old.user_id),
  5.     nvl(new.order_id,old.order_id),
  6.     nvl(new.sku_id,old.sku_id),
  7.     nvl(new.province_id,old.province_id),
  8.     nvl(new.trade_no,old.trade_no),
  9.     nvl(new.out_trade_no,old.out_trade_no),
  10.     nvl(new.payment_type,old.payment_type),
  11.     nvl(new.refund_amount,old.refund_amount),
  12.     nvl(new.refund_status,old.refund_status),
  13.     nvl(new.create_time,old.create_time),
  14.     nvl(new.callback_time,old.callback_time),
  15.     nvl(date_format(nvl(new.callback_time,old.callback_time),'yyyy-MM-dd'),'9999-99-99')
  16. from
  17. (
  18.     select
  19.         id,
  20.         user_id,
  21.         order_id,
  22.         sku_id,
  23.         province_id,
  24.         trade_no,
  25.         out_trade_no,
  26.         payment_type,
  27.         refund_amount,
  28.         refund_status,
  29.         create_time,
  30.         callback_time
  31.     from dwd_refund_payment
  32.     where dt='9999-99-99'
  33. )old
  34. full outer join
  35. (
  36.     select
  37.         rp.id,
  38.         user_id,
  39.         order_id,
  40.         sku_id,
  41.         province_id,
  42.         trade_no,
  43.         out_trade_no,
  44.         payment_type,
  45.         refund_amount,
  46.         refund_status,
  47.         create_time,
  48.         callback_time
  49.     from
  50.     (
  51.         select
  52.             id,
  53.             out_trade_no,
  54.             order_id,
  55.             sku_id,
  56.             payment_type,
  57.             trade_no,
  58.             refund_amount,
  59.             refund_status,
  60.             create_time,
  61.             callback_time
  62.         from ods_refund_payment
  63.         where dt='2020-06-15'
  64.     )rp
  65.     left join
  66.     (
  67.         select
  68.             id,
  69.             user_id,
  70.             province_id
  71.         from ods_order_info
  72.         where dt='2020-06-15'
  73.     )oi
  74.     on rp.order_id=oi.id
  75. )new
  76. on old.id=new.id;

D、查询加载结果

9. 订单事实表 (累积型快照事实表)

A、建表语句

  1. DROP TABLE IF EXISTS dwd_order_info;
  2. CREATE EXTERNAL TABLE dwd_order_info(
  3.     `id` STRING COMMENT '编号',
  4.     `order_status` STRING COMMENT '订单状态',
  5.     `user_id` STRING COMMENT '用户ID',
  6.     `province_id` STRING COMMENT '地区ID',
  7.     `payment_way` STRING COMMENT '支付方式',
  8.     `delivery_address` STRING COMMENT '邮寄地址',
  9.     `out_trade_no` STRING COMMENT '对外交易编号',
  10.     `tracking_no` STRING COMMENT '物流单号',
  11.     `create_time` STRING COMMENT '创建时间(未支付状态)',
  12.     `payment_time` STRING COMMENT '支付时间(已支付状态)',
  13.     `cancel_time` STRING COMMENT '取消时间(已取消状态)',
  14.     `finish_time` STRING COMMENT '完成时间(已完成状态)',
  15.     `refund_time` STRING COMMENT '退款时间(退款中状态)',
  16.     `refund_finish_time` STRING COMMENT '退款完成时间(退款完成状态)',
  17.     `expire_time` STRING COMMENT '过期时间',
  18.     `feight_fee` DECIMAL(16,2) COMMENT '运费',
  19.     `feight_fee_reduce` DECIMAL(16,2) COMMENT '运费减免',
  20.     `activity_reduce_amount` DECIMAL(16,2) COMMENT '活动减免',
  21.     `coupon_reduce_amount` DECIMAL(16,2) COMMENT '优惠券减免',
  22.     `original_amount` DECIMAL(16,2) COMMENT '订单原始价格',
  23.     `final_amount` DECIMAL(16,2) COMMENT '订单最终价格'
  24. ) COMMENT '订单事实表'
  25. PARTITIONED BY (`dt` STRING)
  26. STORED AS PARQUET
  27. LOCATION '/warehouse/gmall/dwd/dwd_order_info/'
  28. TBLPROPERTIES ("parquet.compression"="lzo");

B、分区规划
在这里插入图片描述

C、数据装载

在这里插入图片描述

1.首日装载

  1. insert overwrite table dwd_order_info partition(dt)
  2. select
  3.     oi.id,
  4.     oi.order_status,
  5.     oi.user_id,
  6.     oi.province_id,
  7.     oi.payment_way,
  8.     oi.delivery_address,
  9.     oi.out_trade_no,
  10.     oi.tracking_no,
  11.     oi.create_time,
  12.     times.ts['1002'] payment_time,
  13.     times.ts['1003'] cancel_time,
  14.     times.ts['1004'] finish_time,
  15.     times.ts['1005'] refund_time,
  16.     times.ts['1006'] refund_finish_time,
  17.     oi.expire_time,
  18.     feight_fee,
  19.     feight_fee_reduce,
  20.     activity_reduce_amount,
  21.     coupon_reduce_amount,
  22.     original_amount,
  23.     final_amount,
  24.     case
  25.         when times.ts['1003'] is not null then date_format(times.ts['1003'],'yyyy-MM-dd')
  26.         when times.ts['1004'] is not null and date_add(date_format(times.ts['1004'],'yyyy-MM-dd'),7)<='2020-06-14' and times.ts['1005'] is null then date_add(date_format(times.ts['1004'],'yyyy-MM-dd'),7)
  27.         when times.ts['1006'] is not null then date_format(times.ts['1006'],'yyyy-MM-dd')
  28.         when oi.expire_time is not null then date_format(oi.expire_time,'yyyy-MM-dd')
  29.         else '9999-99-99'
  30.     end
  31. from
  32. (
  33.     select
  34.         *
  35.     from ods_order_info
  36.     where dt='2020-06-14'
  37. )oi
  38. left join
  39. (
  40.     select
  41.         order_id,
  42.         str_to_map(concat_ws(',',collect_set(concat(order_status,'=',operate_time))),',','=') ts
  43.     from ods_order_status_log
  44.     where dt='2020-06-14'
  45.     group by order_id
  46. )times
  47. on oi.id=times.order_id;

2.每日装载

  1. insert overwrite table dwd_order_info partition(dt)
  2. select
  3.     nvl(new.id,old.id),
  4.     nvl(new.order_status,old.order_status),
  5.     nvl(new.user_id,old.user_id),
  6.     nvl(new.province_id,old.province_id),
  7.     nvl(new.payment_way,old.payment_way),
  8.     nvl(new.delivery_address,old.delivery_address),
  9.     nvl(new.out_trade_no,old.out_trade_no),
  10.     nvl(new.tracking_no,old.tracking_no),
  11.     nvl(new.create_time,old.create_time),
  12.     nvl(new.payment_time,old.payment_time),
  13.     nvl(new.cancel_time,old.cancel_time),
  14.     nvl(new.finish_time,old.finish_time),
  15.     nvl(new.refund_time,old.refund_time),
  16.     nvl(new.refund_finish_time,old.refund_finish_time),
  17.     nvl(new.expire_time,old.expire_time),
  18.     nvl(new.feight_fee,old.feight_fee),
  19.     nvl(new.feight_fee_reduce,old.feight_fee_reduce),
  20.     nvl(new.activity_reduce_amount,old.activity_reduce_amount),
  21.     nvl(new.coupon_reduce_amount,old.coupon_reduce_amount),
  22.     nvl(new.original_amount,old.original_amount),
  23.     nvl(new.final_amount,old.final_amount),
  24.     case
  25.         when new.cancel_time is not null then date_format(new.cancel_time,'yyyy-MM-dd')
  26.         when new.finish_time is not null and date_add(date_format(new.finish_time,'yyyy-MM-dd'),7)='2020-06-15' and new.refund_time is null then '2020-06-15'
  27.         when new.refund_finish_time is not null then date_format(new.refund_finish_time,'yyyy-MM-dd')
  28.         when new.expire_time is not null then date_format(new.expire_time,'yyyy-MM-dd')
  29.         else '9999-99-99'
  30.     end
  31. from
  32. (
  33.     select
  34.         id,
  35.         order_status,
  36.         user_id,
  37.         province_id,
  38.         payment_way,
  39.         delivery_address,
  40.         out_trade_no,
  41.         tracking_no,
  42.         create_time,
  43.         payment_time,
  44.         cancel_time,
  45.         finish_time,
  46.         refund_time,
  47.         refund_finish_time,
  48.         expire_time,
  49.         feight_fee,
  50.         feight_fee_reduce,
  51.         activity_reduce_amount,
  52.         coupon_reduce_amount,
  53.         original_amount,
  54.         final_amount
  55.     from dwd_order_info
  56.     where dt='9999-99-99'
  57. )old
  58. full outer join
  59. (
  60.     select
  61.         oi.id,
  62.         oi.order_status,
  63.         oi.user_id,
  64.         oi.province_id,
  65.         oi.payment_way,
  66.         oi.delivery_address,
  67.         oi.out_trade_no,
  68.         oi.tracking_no,
  69.         oi.create_time,
  70.         times.ts['1002'] payment_time,
  71.         times.ts['1003'] cancel_time,
  72.         times.ts['1004'] finish_time,
  73.         times.ts['1005'] refund_time,
  74.         times.ts['1006'] refund_finish_time,
  75.         oi.expire_time,
  76.         feight_fee,
  77.         feight_fee_reduce,
  78.         activity_reduce_amount,
  79.         coupon_reduce_amount,
  80.         original_amount,
  81.         final_amount
  82.     from
  83.     (
  84.         select
  85.             *
  86.         from ods_order_info
  87.         where dt='2020-06-15'
  88.     )oi
  89.     left join
  90.     (
  91.         select
  92.             order_id,
  93.             str_to_map(concat_ws(',',collect_set(concat(order_status,'=',operate_time))),',','=') ts
  94.         from ods_order_status_log
  95.         where dt='2020-06-15'
  96.         group by order_id
  97.     )times
  98.     on oi.id=times.order_id
  99. )new
  100. on old.id=new.id;

10. DWD 层业务数据首日装载脚本

A、编写脚本

在 /home/fancy/bin 目录下创建脚本 ods_to_dwd_db_init.sh

  1. [fancy@node101 bin]$ vim ods_to_dwd_db_init.sh

在脚本中填写如下内容

  1. #!/bin/bash
  2. APP=gmall
  4. if [ -n "$2" ] ;then
  5.    do_date=$2
  6. else 
  7.    echo "请传入日期参数"
  8.    exit
  9. fi 
  11. dwd_order_info="
  12. set hive.exec.dynamic.partition.mode=nonstrict;
  13. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  14. insert overwrite table ${APP}.dwd_order_info partition(dt)
  15. select
  16.     oi.id,
  17.     oi.order_status,
  18.     oi.user_id,
  19.     oi.province_id,
  20.     oi.payment_way,
  21.     oi.delivery_address,
  22.     oi.out_trade_no,
  23.     oi.tracking_no,
  24.     oi.create_time,
  25.     times.ts['1002'] payment_time,
  26.     times.ts['1003'] cancel_time,
  27.     times.ts['1004'] finish_time,
  28.     times.ts['1005'] refund_time,
  29.     times.ts['1006'] refund_finish_time,
  30.     oi.expire_time,
  31.     feight_fee,
  32.     feight_fee_reduce,
  33.     activity_reduce_amount,
  34.     coupon_reduce_amount,
  35.     original_amount,
  36.     final_amount,
  37.     case
  38.         when times.ts['1003'] is not null then date_format(times.ts['1003'],'yyyy-MM-dd')
  39.         when times.ts['1004'] is not null and date_add(date_format(times.ts['1004'],'yyyy-MM-dd'),7)<='$do_date' and times.ts['1005'] is null then date_add(date_format(times.ts['1004'],'yyyy-MM-dd'),7)
  40.         when times.ts['1006'] is not null then date_format(times.ts['1006'],'yyyy-MM-dd')
  41.         when oi.expire_time is not null then date_format(oi.expire_time,'yyyy-MM-dd')
  42.         else '9999-99-99'
  43.     end
  44. from
  45. (
  46.     select
  47.         *
  48.     from ${APP}.ods_order_info
  49.     where dt='$do_date'
  50. )oi
  51. left join
  52. (
  53.     select
  54.         order_id,
  55.         str_to_map(concat_ws(',',collect_set(concat(order_status,'=',operate_time))),',','=') ts
  56.     from ${APP}.ods_order_status_log
  57.     where dt='$do_date'
  58.     group by order_id
  59. )times
  60. on oi.id=times.order_id;"
  62. dwd_order_detail="
  63. set hive.exec.dynamic.partition.mode=nonstrict;
  64. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  65. insert overwrite table ${APP}.dwd_order_detail partition(dt)
  66. select
  67.     od.id,
  68.     od.order_id,
  69.     oi.user_id,
  70.     od.sku_id,
  71.     oi.province_id,
  72.     oda.activity_id,
  73.     oda.activity_rule_id,
  74.     odc.coupon_id,
  75.     od.create_time,
  76.     od.source_type,
  77.     od.source_id,
  78.     od.sku_num,
  79.     od.order_price*od.sku_num,
  80.     od.split_activity_amount,
  81.     od.split_coupon_amount,
  82.     od.split_final_amount,
  83.     date_format(create_time,'yyyy-MM-dd')
  84. from
  85. (
  86.     select
  87.         *
  88.     from ${APP}.ods_order_detail
  89.     where dt='$do_date'
  90. )od
  91. left join
  92. (
  93.     select
  94.         id,
  95.         user_id,
  96.         province_id
  97.     from ${APP}.ods_order_info
  98.     where dt='$do_date'
  99. )oi
  100. on od.order_id=oi.id
  101. left join
  102. (
  103.     select
  104.         order_detail_id,
  105.         activity_id,
  106.         activity_rule_id
  107.     from ${APP}.ods_order_detail_activity
  108.     where dt='$do_date'
  109. )oda
  110. on od.id=oda.order_detail_id
  111. left join
  112. (
  113.     select
  114.         order_detail_id,
  115.         coupon_id
  116.     from ${APP}.ods_order_detail_coupon
  117.     where dt='$do_date'
  118. )odc
  119. on od.id=odc.order_detail_id;"
  121. dwd_payment_info="
  122. set hive.exec.dynamic.partition.mode=nonstrict;
  123. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  124. insert overwrite table ${APP}.dwd_payment_info partition(dt)
  125. select
  126.     pi.id,
  127.     pi.order_id,
  128.     pi.user_id,
  129.     oi.province_id,
  130.     pi.trade_no,
  131.     pi.out_trade_no,
  132.     pi.payment_type,
  133.     pi.payment_amount,
  134.     pi.payment_status,
  135.     pi.create_time,
  136.     pi.callback_time,
  137.     nvl(date_format(pi.callback_time,'yyyy-MM-dd'),'9999-99-99')
  138. from
  139. (
  140.     select * from ${APP}.ods_payment_info where dt='$do_date'
  141. )pi
  142. left join
  143. (
  144.     select id,province_id from ${APP}.ods_order_info where dt='$do_date'
  145. )oi
  146. on pi.order_id=oi.id;"
  148. dwd_cart_info="
  149. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  150. insert overwrite table ${APP}.dwd_cart_info partition(dt='$do_date')
  151. select
  152.     id,
  153.     user_id,
  154.     sku_id,
  155.     source_type,
  156.     source_id,
  157.     cart_price,
  158.     is_ordered,
  159.     create_time,
  160.     operate_time,
  161.     order_time,
  162.     sku_num
  163. from ${APP}.ods_cart_info
  164. where dt='$do_date';"
  166. dwd_comment_info="
  167. set hive.exec.dynamic.partition.mode=nonstrict;
  168. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  169. insert overwrite table ${APP}.dwd_comment_info partition(dt)
  170. select
  171.     id,
  172.     user_id,
  173.     sku_id,
  174.     spu_id,
  175.     order_id,
  176.     appraise,
  177.     create_time,
  178.     date_format(create_time,'yyyy-MM-dd')
  179. from ${APP}.ods_comment_info
  180. where dt='$do_date';
  181. "
  183. dwd_favor_info="
  184. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  185. insert overwrite table ${APP}.dwd_favor_info partition(dt='$do_date')
  186. select
  187.     id,
  188.     user_id,
  189.     sku_id,
  190.     spu_id,
  191.     is_cancel,
  192.     create_time,
  193.     cancel_time
  194. from ${APP}.ods_favor_info
  195. where dt='$do_date';"
  197. dwd_coupon_use="
  198. set hive.exec.dynamic.partition.mode=nonstrict;
  199. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  200. insert overwrite table ${APP}.dwd_coupon_use partition(dt)
  201. select
  202.     id,
  203.     coupon_id,
  204.     user_id,
  205.     order_id,
  206.     coupon_status,
  207.     get_time,
  208.     using_time,
  209.     used_time,
  210.     expire_time,
  211.     coalesce(date_format(used_time,'yyyy-MM-dd'),date_format(expire_time,'yyyy-MM-dd'),'9999-99-99')
  212. from ${APP}.ods_coupon_use
  213. where dt='$do_date';"
  215. dwd_order_refund_info="
  216. set hive.exec.dynamic.partition.mode=nonstrict;
  217. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  218. insert overwrite table ${APP}.dwd_order_refund_info partition(dt)
  219. select
  220.     ri.id,
  221.     ri.user_id,
  222.     ri.order_id,
  223.     ri.sku_id,
  224.     oi.province_id,
  225.     ri.refund_type,
  226.     ri.refund_num,
  227.     ri.refund_amount,
  228.     ri.refund_reason_type,
  229.     ri.create_time,
  230.     date_format(ri.create_time,'yyyy-MM-dd')
  231. from
  232. (
  233.     select * from ${APP}.ods_order_refund_info where dt='$do_date'
  234. )ri
  235. left join
  236. (
  237.     select id,province_id from ${APP}.ods_order_info where dt='$do_date'
  238. )oi
  239. on ri.order_id=oi.id;"
  241. dwd_refund_payment="
  242. set hive.exec.dynamic.partition.mode=nonstrict;
  243. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  244. insert overwrite table ${APP}.dwd_refund_payment partition(dt)
  245. select
  246.     rp.id,
  247.     user_id,
  248.     order_id,
  249.     sku_id,
  250.     province_id,
  251.     trade_no,
  252.     out_trade_no,
  253.     payment_type,
  254.     refund_amount,
  255.     refund_status,
  256.     create_time,
  257.     callback_time,
  258.     nvl(date_format(callback_time,'yyyy-MM-dd'),'9999-99-99')
  259. from
  260. (
  261.     select
  262.         id,
  263.         out_trade_no,
  264.         order_id,
  265.         sku_id,
  266.         payment_type,
  267.         trade_no,
  268.         refund_amount,
  269.         refund_status,
  270.         create_time,
  271.         callback_time
  272.     from ${APP}.ods_refund_payment
  273.     where dt='$do_date'
  274. )rp
  275. left join
  276. (
  277.     select
  278.         id,
  279.         user_id,
  280.         province_id
  281.     from ${APP}.ods_order_info
  282.     where dt='$do_date'
  283. )oi
  284. on rp.order_id=oi.id;"
  286. case $1 in
  287.     dwd_order_info )
  288.         hive -e "$dwd_order_info"
  289.     ;;
  290.     dwd_order_detail )
  291.         hive -e "$dwd_order_detail"
  292.     ;;
  293.     dwd_payment_info )
  294.         hive -e "$dwd_payment_info"
  295.     ;;
  296.     dwd_cart_info )
  297.         hive -e "$dwd_cart_info"
  298.     ;;
  299.     dwd_comment_info )
  300.         hive -e "$dwd_comment_info"
  301.     ;;
  302.     dwd_favor_info )
  303.         hive -e "$dwd_favor_info"
  304.     ;;
  305.     dwd_coupon_use )
  306.         hive -e "$dwd_coupon_use"
  307.     ;;
  308.     dwd_order_refund_info )
  309.         hive -e "$dwd_order_refund_info"
  310.     ;;
  311.     dwd_refund_payment )
  312.         hive -e "$dwd_refund_payment"
  313.     ;;
  314.     all )
  315.         hive -e "$dwd_order_info$dwd_order_detail$dwd_payment_info$dwd_cart_info$dwd_comment_info$dwd_favor_info$dwd_coupon_use$dwd_order_refund_info$dwd_refund_payment"
  316.     ;;
  317. esac

增加执行权限

  1. [fancy@node101 bin]$ chmod +x ods_to_dwd_db_init.sh

B、脚本使用

执行脚本

  1. [fancy@node101 bin]$ ods_to_dwd_db_init.sh all 2020-06-14

查看数据是否导入成功

11. DWD层业务数据每日装载脚本

A、编写脚本

在 /home/fancy/bin 目录下创建脚本ods_to_dwd_db.sh

  1. [fancy@node101 bin]$ vim ods_to_dwd_db.sh

在脚本中填写如下内容

  1. #!/bin/bash
  3. APP=gmall
  4. # 如果是输入的日期按照取输入日期；如果没输入日期取当前时间的前一天
  5. if [ -n "$2" ] ;then
  6.     do_date=$2
  7. else 
  8.     do_date=`date -d "-1 day" +%F`
  9. fi
  12. # 假设某累积型快照事实表，某天所有的业务记录全部完成，则会导致9999-99-99分区的数据未被覆盖，从而导致数据重复，该函数根据9999-99-99分区的数据的末次修改时间判断其是否被覆盖了，如果未被覆盖，就手动清理
  13. clear_data(){
  14.     current_date=`date +%F`
  15.     current_date_timestamp=`date -d "$current_date" +%s`
  17.     last_modified_date=`hadoop fs -ls /warehouse/gmall/dwd/$1 | grep '9999-99-99' | awk '{print $6}'`
  18.     last_modified_date_timestamp=`date -d "$last_modified_date" +%s`
  20.     if [[ $last_modified_date_timestamp -lt $current_date_timestamp ]]; then
  21.         echo "clear table $1 partition(dt=9999-99-99)"
  22.         hadoop fs -rm -r -f /warehouse/gmall/dwd/$1/dt=9999-99-99/*
  23.     fi
  24. }
  26. dwd_order_info="
  27. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  28. set hive.exec.dynamic.partition.mode=nonstrict;
  29. insert overwrite table ${APP}.dwd_order_info partition(dt)
  30. select
  31.     nvl(new.id,old.id),
  32.     nvl(new.order_status,old.order_status),
  33.     nvl(new.user_id,old.user_id),
  34.     nvl(new.province_id,old.province_id),
  35.     nvl(new.payment_way,old.payment_way),
  36.     nvl(new.delivery_address,old.delivery_address),
  37.     nvl(new.out_trade_no,old.out_trade_no),
  38.     nvl(new.tracking_no,old.tracking_no),
  39.     nvl(new.create_time,old.create_time),
  40.     nvl(new.payment_time,old.payment_time),
  41.     nvl(new.cancel_time,old.cancel_time),
  42.     nvl(new.finish_time,old.finish_time),
  43.     nvl(new.refund_time,old.refund_time),
  44.     nvl(new.refund_finish_time,old.refund_finish_time),
  45.     nvl(new.expire_time,old.expire_time),
  46.     nvl(new.feight_fee,old.feight_fee),
  47.     nvl(new.feight_fee_reduce,old.feight_fee_reduce),
  48.     nvl(new.activity_reduce_amount,old.activity_reduce_amount),
  49.     nvl(new.coupon_reduce_amount,old.coupon_reduce_amount),
  50.     nvl(new.original_amount,old.original_amount),
  51.     nvl(new.final_amount,old.final_amount),
  52.     case
  53.         when new.cancel_time is not null then date_format(new.cancel_time,'yyyy-MM-dd')
  54.         when new.finish_time is not null and date_add(date_format(new.finish_time,'yyyy-MM-dd'),7)='$do_date' and new.refund_time is null then '$do_date'
  55.         when new.refund_finish_time is not null then date_format(new.refund_finish_time,'yyyy-MM-dd')
  56.         when new.expire_time is not null then date_format(new.expire_time,'yyyy-MM-dd')
  57.         else '9999-99-99'
  58.     end
  59. from
  60. (
  61.     select
  62.         id,
  63.         order_status,
  64.         user_id,
  65.         province_id,
  66.         payment_way,
  67.         delivery_address,
  68.         out_trade_no,
  69.         tracking_no,
  70.         create_time,
  71.         payment_time,
  72.         cancel_time,
  73.         finish_time,
  74.         refund_time,
  75.         refund_finish_time,
  76.         expire_time,
  77.         feight_fee,
  78.         feight_fee_reduce,
  79.         activity_reduce_amount,
  80.         coupon_reduce_amount,
  81.         original_amount,
  82.         final_amount
  83.     from ${APP}.dwd_order_info
  84.     where dt='9999-99-99'
  85. )old
  86. full outer join
  87. (
  88.     select
  89.         oi.id,
  90.         oi.order_status,
  91.         oi.user_id,
  92.         oi.province_id,
  93.         oi.payment_way,
  94.         oi.delivery_address,
  95.         oi.out_trade_no,
  96.         oi.tracking_no,
  97.         oi.create_time,
  98.         times.ts['1002'] payment_time,
  99.         times.ts['1003'] cancel_time,
  100.         times.ts['1004'] finish_time,
  101.         times.ts['1005'] refund_time,
  102.         times.ts['1006'] refund_finish_time,
  103.         oi.expire_time,
  104.         feight_fee,
  105.         feight_fee_reduce,
  106.         activity_reduce_amount,
  107.         coupon_reduce_amount,
  108.         original_amount,
  109.         final_amount
  110.     from
  111.     (
  112.         select
  113.             *
  114.         from ${APP}.ods_order_info
  115.         where dt='$do_date'
  116.     )oi
  117.     left join
  118.     (
  119.         select
  120.             order_id,
  121.             str_to_map(concat_ws(',',collect_set(concat(order_status,'=',operate_time))),',','=') ts
  122.         from ${APP}.ods_order_status_log
  123.         where dt='$do_date'
  124.         group by order_id
  125.     )times
  126.     on oi.id=times.order_id
  127. )new
  128. on old.id=new.id;"
  130. dwd_order_detail="
  131. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  132. insert overwrite table ${APP}.dwd_order_detail partition(dt='$do_date')
  133. select
  134.     od.id,
  135.     od.order_id,
  136.     oi.user_id,
  137.     od.sku_id,
  138.     oi.province_id,
  139.     oda.activity_id,
  140.     oda.activity_rule_id,
  141.     odc.coupon_id,
  142.     od.create_time,
  143.     od.source_type,
  144.     od.source_id,
  145.     od.sku_num,
  146.     od.order_price*od.sku_num,
  147.     od.split_activity_amount,
  148.     od.split_coupon_amount,
  149.     od.split_final_amount
  150. from
  151. (
  152.     select
  153.         *
  154.     from ${APP}.ods_order_detail
  155.     where dt='$do_date'
  156. )od
  157. left join
  158. (
  159.     select
  160.         id,
  161.         user_id,
  162.         province_id
  163.     from ${APP}.ods_order_info
  164.     where dt='$do_date'
  165. )oi
  166. on od.order_id=oi.id
  167. left join
  168. (
  169.     select
  170.         order_detail_id,
  171.         activity_id,
  172.         activity_rule_id
  173.     from ${APP}.ods_order_detail_activity
  174.     where dt='$do_date'
  175. )oda
  176. on od.id=oda.order_detail_id
  177. left join
  178. (
  179.     select
  180.         order_detail_id,
  181.         coupon_id
  182.     from ${APP}.ods_order_detail_coupon
  183.     where dt='$do_date'
  184. )odc
  185. on od.id=odc.order_detail_id;"
  188. dwd_payment_info="
  189. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  190. set hive.exec.dynamic.partition.mode=nonstrict;
  191. insert overwrite table ${APP}.dwd_payment_info partition(dt)
  192. select
  193.     nvl(new.id,old.id),
  194.     nvl(new.order_id,old.order_id),
  195.     nvl(new.user_id,old.user_id),
  196.     nvl(new.province_id,old.province_id),
  197.     nvl(new.trade_no,old.trade_no),
  198.     nvl(new.out_trade_no,old.out_trade_no),
  199.     nvl(new.payment_type,old.payment_type),
  200.     nvl(new.payment_amount,old.payment_amount),
  201.     nvl(new.payment_status,old.payment_status),
  202.     nvl(new.create_time,old.create_time),
  203.     nvl(new.callback_time,old.callback_time),
  204.     nvl(date_format(nvl(new.callback_time,old.callback_time),'yyyy-MM-dd'),'9999-99-99')
  205. from
  206. (
  207.     select id,
  208.        order_id,
  209.        user_id,
  210.        province_id,
  211.        trade_no,
  212.        out_trade_no,
  213.        payment_type,
  214.        payment_amount,
  215.        payment_status,
  216.        create_time,
  217.        callback_time
  218.     from ${APP}.dwd_payment_info
  219.     where dt = '9999-99-99'
  220. )old
  221. full outer join
  222. (
  223.     select
  224.         pi.id,
  225.         pi.out_trade_no,
  226.         pi.order_id,
  227.         pi.user_id,
  228.         oi.province_id,
  229.         pi.payment_type,
  230.         pi.trade_no,
  231.         pi.payment_amount,
  232.         pi.payment_status,
  233.         pi.create_time,
  234.         pi.callback_time
  235.     from
  236.     (
  237.         select * from ${APP}.ods_payment_info where dt='$do_date'
  238.     )pi
  239.     left join
  240.     (
  241.         select id,province_id from ${APP}.ods_order_info where dt='$do_date'
  242.     )oi
  243.     on pi.order_id=oi.id
  244. )new
  245. on old.id=new.id;"
  247. dwd_cart_info="
  248. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  249. insert overwrite table ${APP}.dwd_cart_info partition(dt='$do_date')
  250. select
  251.     id,
  252.     user_id,
  253.     sku_id,
  254.     source_type,
  255.     source_id,
  256.     cart_price,
  257.     is_ordered,
  258.     create_time,
  259.     operate_time,
  260.     order_time,
  261.     sku_num
  262. from ${APP}.ods_cart_info
  263. where dt='$do_date';"
  266. dwd_comment_info="
  267. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  268. insert overwrite table ${APP}.dwd_comment_info partition(dt='$do_date')
  269. select
  270.     id,
  271.     user_id,
  272.     sku_id,
  273.     spu_id,
  274.     order_id,
  275.     appraise,
  276.     create_time
  277. from ${APP}.ods_comment_info where dt='$do_date';"
  280. dwd_favor_info="
  281. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  282. insert overwrite table ${APP}.dwd_favor_info partition(dt='$do_date')
  283. select
  284.     id,
  285.     user_id,
  286.     sku_id,
  287.     spu_id,
  288.     is_cancel,
  289.     create_time,
  290.     cancel_time
  291. from ${APP}.ods_favor_info
  292. where dt='$do_date';"
  295. dwd_coupon_use="
  296. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  297. set hive.exec.dynamic.partition.mode=nonstrict;
  298. insert overwrite table ${APP}.dwd_coupon_use partition(dt)
  299. select
  300.     nvl(new.id,old.id),
  301.     nvl(new.coupon_id,old.coupon_id),
  302.     nvl(new.user_id,old.user_id),
  303.     nvl(new.order_id,old.order_id),
  304.     nvl(new.coupon_status,old.coupon_status),
  305.     nvl(new.get_time,old.get_time),
  306.     nvl(new.using_time,old.using_time),
  307.     nvl(new.used_time,old.used_time),
  308.     nvl(new.expire_time,old.expire_time),
  309.     coalesce(date_format(nvl(new.used_time,old.used_time),'yyyy-MM-dd'),date_format(nvl(new.expire_time,old.expire_time),'yyyy-MM-dd'),'9999-99-99')
  310. from
  311. (
  312.     select
  313.         id,
  314.         coupon_id,
  315.         user_id,
  316.         order_id,
  317.         coupon_status,
  318.         get_time,
  319.         using_time,
  320.         used_time,
  321.         expire_time
  322.     from ${APP}.dwd_coupon_use
  323.     where dt='9999-99-99'
  324. )old
  325. full outer join
  326. (
  327.     select
  328.         id,
  329.         coupon_id,
  330.         user_id,
  331.         order_id,
  332.         coupon_status,
  333.         get_time,
  334.         using_time,
  335.         used_time,
  336.         expire_time
  337.     from ${APP}.ods_coupon_use
  338.     where dt='$do_date'
  339. )new
  340. on old.id=new.id;"
  342. dwd_order_refund_info="
  343. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  344. insert overwrite table ${APP}.dwd_order_refund_info partition(dt='$do_date')
  345. select
  346.     ri.id,
  347.     ri.user_id,
  348.     ri.order_id,
  349.     ri.sku_id,
  350.     oi.province_id,
  351.     ri.refund_type,
  352.     ri.refund_num,
  353.     ri.refund_amount,
  354.     ri.refund_reason_type,
  355.     ri.create_time
  356. from
  357. (
  358.     select * from ${APP}.ods_order_refund_info where dt='$do_date'
  359. )ri
  360. left join
  361. (
  362.     select id,province_id from ${APP}.ods_order_info where dt='$do_date'
  363. )oi
  364. on ri.order_id=oi.id;"
  367. dwd_refund_payment="
  368. set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
  369. set hive.exec.dynamic.partition.mode=nonstrict;
  370. insert overwrite table ${APP}.dwd_refund_payment partition(dt)
  371. select
  372.     nvl(new.id,old.id),
  373.     nvl(new.user_id,old.user_id),
  374.     nvl(new.order_id,old.order_id),
  375.     nvl(new.sku_id,old.sku_id),
  376.     nvl(new.province_id,old.province_id),
  377.     nvl(new.trade_no,old.trade_no),
  378.     nvl(new.out_trade_no,old.out_trade_no),
  379.     nvl(new.payment_type,old.payment_type),
  380.     nvl(new.refund_amount,old.refund_amount),
  381.     nvl(new.refund_status,old.refund_status),
  382.     nvl(new.create_time,old.create_time),
  383.     nvl(new.callback_time,old.callback_time),
  384.     nvl(date_format(nvl(new.callback_time,old.callback_time),'yyyy-MM-dd'),'9999-99-99')
  385. from
  386. (
  387.     select
  388.         id,
  389.         user_id,
  390.         order_id,
  391.         sku_id,
  392.         province_id,
  393.         trade_no,
  394.         out_trade_no,
  395.         payment_type,
  396.         refund_amount,
  397.         refund_status,
  398.         create_time,
  399.         callback_time
  400.     from ${APP}.dwd_refund_payment
  401.     where dt='9999-99-99'
  402. )old
  403. full outer join
  404. (
  405.     select
  406.         rp.id,
  407.         user_id,
  408.         order_id,
  409.         sku_id,
  410.         province_id,
  411.         trade_no,
  412.         out_trade_no,
  413.         payment_type,
  414.         refund_amount,
  415.         refund_status,
  416.         create_time,
  417.         callback_time
  418.     from
  419.     (
  420.         select
  421.             id,
  422.             out_trade_no,
  423.             order_id,
  424.             sku_id,
  425.             payment_type,
  426.             trade_no,
  427.             refund_amount,
  428.             refund_status,
  429.             create_time,
  430.             callback_time
  431.         from ${APP}.ods_refund_payment
  432.         where dt='$do_date'
  433.     )rp
  434.     left join
  435.     (
  436.         select
  437.             id,
  438.             user_id,
  439.             province_id
  440.         from ${APP}.ods_order_info
  441.         where dt='$do_date'
  442.     )oi
  443.     on rp.order_id=oi.id
  444. )new
  445. on old.id=new.id;"
  447. case $1 in
  448.     dwd_order_info )
  449.         hive -e "$dwd_order_info"
  450.         clear_data dwd_order_info
  451.     ;;
  452.     dwd_order_detail )
  453.         hive -e "$dwd_order_detail"
  454.     ;;
  455.     dwd_payment_info )
  456.         hive -e "$dwd_payment_info"
  457.         clear_data dwd_payment_info
  458.     ;;
  459.     dwd_cart_info )
  460.         hive -e "$dwd_cart_info"
  461.     ;;
  462.     dwd_comment_info )
  463.         hive -e "$dwd_comment_info"
  464.     ;;
  465.     dwd_favor_info )
  466.         hive -e "$dwd_favor_info"
  467.     ;;
  468.     dwd_coupon_use )
  469.         hive -e "$dwd_coupon_use"
  470.         clear_data dwd_coupon_use
  471.     ;;
  472.     dwd_order_refund_info )
  473.         hive -e "$dwd_order_refund_info"
  474.     ;;
  475.     dwd_refund_payment )
  476.         hive -e "$dwd_refund_payment"
  477.         clear_data dwd_refund_payment
  478.     ;;
  479.     all )
  480.         hive -e "$dwd_order_info$dwd_order_detail$dwd_payment_info$dwd_cart_info$dwd_comment_info$dwd_favor_info$dwd_coupon_use$dwd_order_refund_info$dwd_refund_payment"
  481.         clear_data dwd_order_info
  482.         clear_data dwd_payment_info
  483.         clear_data dwd_coupon_use
  484.         clear_data dwd_refund_payment
  485.     ;;
  486. esac

增加脚本执行权限

  1. [fancy@node101 bin]$ chmod 777 ods_to_dwd_db.sh

B、脚本使用

执行脚本

  1. [fancy@node101 bin]$ ods_to_dwd_db.sh all 2020-06-14

查看数据是否导入成功

发表评论

表情：
评论列表 （有 0 条评论，153人围观）

还没有评论，来说两句吧...

相关阅读