大数据-案例-离线数仓-电商：【MySQL(业务)-ETL(Kettle)】+【前端JS埋点-＞日志-＞Flume-＞HDFS-＞ETL(SparkRDD)】-＞Hive数仓-＞MySQL-＞可视化_电商项目数据etl

一、电商行业简介

1、电商行业分析

近年来，中国的电子商务快速发展，交易额连创新高，电子商务在各领域的应用不断拓展和深化、相关服务业蓬勃发展、支撑体系不断健全完善、创新的动力和能力 不断增强。电子商务正在与实体经济深度融合，进入规模性发展阶段，对经济社会生活的影响不断增大，正成为我国经济发展的新引擎。

中国电子商务研究中心数据显示，截止到 2012 年底，中国电子商务市场交易规模达 7.85万亿人民币，同比增长 30.83%。其中，B2B 电子商务交易额 达 6.25 万亿，同比增长 27%。而 2011 年全年，中国电子商务市场交易额达 6 万亿人民币，同比增长 33%，占 GDP 比重上升到 13%；2012 年，电子商务占 GDP 的比重已经高达 15%。

电商行业技术特点

• 技术新

• 技术范围广

• 分布式

• 高并发、集群、负载均衡(Nginx)、高可用(备机)

• 海量数据

• 业务复杂

• 系统安全

2、电商业务系统简介

品优购网上商城是一个综合性的 B2B2C 平台，类似京东商城、天猫商城。网站采用商家入驻的模式，商家入驻平台提交申请，有平台进行资质审核，审核通过后，商家拥有独立的管理后台录入商品信息。商品经过平台审核后即可发布。

品优购网上商城主要分为

• 网站前台
• 运营商后台
• 商家管理后台

2.1 网站前台

主要包括

• 网站首页
• 商家首页
• 商品详细页
• 搜索页
• 会员中心
• 订单与支付相关页面
• 秒杀频道等。

2.2 运营商后台

是运营商的运营人员的管理后台。 主要包括商家审核、品牌管理、规格管理、模板管理、商品分类管理、商品审核、广告类型管理、广告管理、订单查询、商家结算等。

2.3 商家管理后台

入驻的商家进行管理的后台，主要功能是对商品的管理以及订单查询统计、资金结算等功能。

二、大数据数仓项目简介

1、业务流程

本次数仓业务流程主要分为两类，

• 一类是用户下单、提交订单、支付、退款这一条线，
• 另一类是我们收集用户的页面行为数据：用户搜索商品、添加购物车 、提交订单、支付订单 的日志数据，分析电商网站常见的PV，UV，GMV，

GMV (Gross Merchandise Volume)：主要是指网站的成交金额，而这里的成交金额包括：付款金额和未付款。

千亿级数仓模仿阿里巴巴双十一的大屏显示功能实现的互联网电商指标的离线分析，同时也模仿了阿里巴巴大数据平台上面数据仓库的设计思想和理念。通过这个项目，能够掌握以下三个核心技能：

1、数据仓库的概念和建设过程

2、离线数据仓库的功能、使用场景和常用的技术栈

2、大数据离线数仓项目架构

3、项目具体技术简介

• Kettle
• 缓慢变化维(拉链表):时间维度，脚本生成，时间维度生成之后不会变化，SCD问题我们使用拉链表来解决；
• Hive
• kettle：导出数据的工具
• Spark SQL：计算引擎
• Kylin：计算引擎，进行预计算之后的多维统计分析可以达到亚秒级别。

4、项目环境介绍

• 业务数据量

  • 用户数：300W
  • 每日订单量：10W
  • 每日交易额：700W
  • 商家数：5W
  • 商品数：45W
  • PV：500W
  • UV：50W

• 数据在hdfs中平均每天 40G左右的速度增长，存储3份，每天增长大概120G,存储hive表时

  会说过parquet格式+snappy压缩

• 硬件资源

  • 数量：30台

  • CPU资源：24核

  • 内存：128G

  • 硬盘：4T

5、业务系统表结构介绍

订单表itcast_orders

订单明细表 itcast_order_goods

商品信息表 itcast_goods

店铺表 itcast_shops

商品分类表 itcast_goods_cats

组织结构表 itcast_org

订单退货表 itcast_order_refunds

用户表 itcast_users

用户收货地址表 itcast_user_address

支付方式表 itcast_payments

6、项目环境初始化

三、业务数据采集与处理

1、导入mysql模拟数据

将资料\mysql建表语句\10tables.sql文件上传到linux，登录mysql使用source命令执行该sql文件创建数据库和表

mysql -uroot -p
source /root/sql/10tables.sql;
1
2

2、Hive分层说明

• 分库存放

  • ods层
  • dw层
  • ads层

• 命名规则

  • ods层表与原始数据库表名称相同

  • dw层表

    • fact_前缀表示事实表

    • dim_前缀表示维度表

创建分层数据库：

create database itcast_ods;
create database itcast_dw;
create database itcast_ads;
1
2
3

3、创建ods层数据表

• hive 分为外部表与内部表，为便于管理，该部分均使用内部表
• 执行资料\hiveods层建表语句\ods_create_table.sql

  -- 创建ods层订单表
  drop table if exists `itcast_ods`.`itcast_orders`;
  create EXTERNAL table `itcast_ods`.`itcast_orders`(
      orderId            bigint,
      orderNo            string,
      shopId             bigint,
      userId             bigint,
      orderStatus        bigint,
      goodsMoney         double,
      deliverType        bigint,
      deliverMoney       double,
      totalMoney         double,
      realTotalMoney     double,
      payType            bigint,
      isPay              bigint,
      areaId             bigint,
      userAddressId      bigint,
      areaIdPath         string,
      userName           string,
      userAddress        string,
      userPhone          string,
      orderScore         bigint,
      isInvoice          bigint,
      invoiceClient      string,
      orderRemarks       string,
      orderSrc           bigint,
      needPay            double,
      payRand            bigint,
      orderType          bigint,
      isRefund           bigint,
      isAppraise         bigint,
      cancelReason       bigint,
      rejectReason       bigint,
      rejectOtherReason  string,
      isClosed           bigint,
      goodsSearchKeys    string,
      orderunique        string,
      receiveTime        string,
      deliveryTime       string,
      tradeNo            string,
      dataFlag           bigint,
      createTime         string,
      settlementId       bigint,
      commissionFee      double,
      scoreMoney         double,
      useScore           bigint,
      orderCode          string,
      extraJson          string,
      orderCodeTargetId  bigint,
      noticeDeliver      bigint,
      invoiceJson        string,
      lockCashMoney      double,
      payTime            string,
      isBatch            bigint,
      totalPayFee        bigint,
      modifiedTime        string
  )
  partitioned by (dt string)
  STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
  
  -- 创建ods层订单明细表
  drop table if exists `itcast_ods`.`itcast_order_goods`;
  create EXTERNAL table `itcast_ods`.`itcast_order_goods`(
      ogId            bigint,
      orderId         bigint,
      goodsId         bigint,
      goodsNum        bigint,
      goodsPrice      double,
      payPrice        double,
      goodsSpecId     bigint,
      goodsSpecNames  string,
      goodsName       string,
      goodsImg        string,
      extraJson       string,
      goodsType       bigint,
      commissionRate  double,
      goodsCode       string,
      promotionJson   string,
      createtime      string
  )
  partitioned by (dt string)
  STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
  
  -- 创建ods层店铺表
  drop table if exists `itcast_ods`.`itcast_shops`;
  create EXTERNAL table `itcast_ods`.`itcast_shops`(
      shopId             bigint,
      shopSn             string,
      userId             bigint,
      areaIdPath         string,
      areaId             bigint,
      isSelf             bigint,
      shopName           string,
      shopkeeper         string,
      telephone          string,
      shopCompany        string,
      shopImg            string,
      shopTel            string,
      shopQQ             string,
      shopWangWang       string,
      shopAddress        string,
      bankId             bigint,
      bankNo             string,
      bankUserName       string,
      isInvoice          bigint,
      invoiceRemarks     string,
      serviceStartTime   bigint,
      serviceEndTime     bigint,
      freight            bigint,
      shopAtive          bigint,
      shopStatus         bigint,
      statusDesc         string,
      dataFlag           bigint,
      createTime         string,
      shopMoney          double,
      lockMoney          double,
      noSettledOrderNum  bigint,
      noSettledOrderFee  double,
      paymentMoney       double,
      bankAreaId         bigint,
      bankAreaIdPath     string,
      applyStatus        bigint,
      applyDesc          string,
      applyTime          string,
      applyStep          bigint,
      shopNotice         string,
      rechargeMoney      double,
      longitude          double,
      latitude           double,
      mapLevel           bigint,
      BDcode             string,
      modifyTime         string
  )
  partitioned by (dt string)
  STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
  
  -- 创建ods层商品表
  drop table if exists `itcast_ods`.`itcast_goods`;
  create EXTERNAL table `itcast_ods`.`itcast_goods`(
      goodsId              bigint,
      goodsSn              string,
      productNo            string,
      goodsName            string,
      goodsImg             string,
      shopId               bigint,
      goodsType            bigint,
      marketPrice          double,
      shopPrice            double,
      warnStock            bigint,
      goodsStock           bigint,
      goodsUnit            string,
      goodsTips            string,
      isSale               bigint,
      isBest               bigint,
      isHot                bigint,
      isNew                bigint,
      isRecom              bigint,
      goodsCatIdPath       string,
      goodsCatId           bigint,
      shopCatId1           bigint,
      shopCatId2           bigint,
      brandId              bigint,
      goodsDesc            string,
      goodsStatus          bigint,
      saleNum              bigint,
      saleTime             string,
      visitNum             bigint,
      appraiseNum          bigint,
      isSpec               bigint,
      gallery              string,
      goodsSeoKeywords     string,
      illegalRemarks       string,
      dataFlag             bigint,
      createTime           string,
      isFreeShipping       bigint,
      goodsSerachKeywords  string,
      modifyTime           string
  )
  partitioned by (dt string)
  STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
  
  -- 创建ods层组织机构表
  drop table `itcast_ods`.`itcast_org`;
  create EXTERNAL table `itcast_ods`.`itcast_org`(
      orgId        bigint,
      parentId     bigint,
      orgName      string,
      orgLevel     bigint,
      managerCode  string,
      isdelete     bigint,
      createTime   string,
      updateTime   string,
      isShow       bigint,
      orgType      bigint
  )
  partitioned by (dt string)
  STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
  
  -- 创建ods层商品分类表
  drop table if exists `itcast_ods`.`itcast_goods_cats`;
  create EXTERNAL table `itcast_ods`.`itcast_goods_cats`(
      catId               bigint,
      parentId            bigint,
      catName             string,
      isShow              bigint,
      isFloor             bigint,
      catSort             bigint,
      dataFlag            bigint,
      createTime          string,
      commissionRate      double,
      catImg              string,
      subTitle            string,
      simpleName          string,
      seoTitle            string,
      seoKeywords         string,
      seoDes              string,
      catListTheme        string,
      detailTheme         string,
      mobileCatListTheme  string,
      mobileDetailTheme   string,
      wechatCatListTheme  string,
      wechatDetailTheme   string,
      cat_level           bigint
  )
  partitioned by (dt string)
  STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
  
  -- 创建ods层用户表
  drop table if exists `itcast_ods`.`itcast_users`;
  create EXTERNAL table `itcast_ods`.`itcast_users`(
      userId          bigint,
      loginName       string,
      loginSecret     bigint,
      loginPwd        string,
      userType        bigint,
      userSex         bigint,
      userName        string,
      trueName        string,
      brithday        string,
      userPhoto       string,
      userQQ          string,
      userPhone       string,
      userEmail       string,
      userScore       bigint,
      userTotalScore  bigint,
      lastIP          string,
      lastTime        string,
      userFrom        bigint,
      userMoney       double,
      lockMoney       double,
      userStatus      bigint,
      dataFlag        bigint,
      createTime      string,
      payPwd          string,
      rechargeMoney   double,
      isInform        bigint
  )
  partitioned by (dt string)
  STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
  
  -- 创建ods层退货表
  drop table if exists `itcast_ods`.`itcast_order_refunds`;
  create EXTERNAL table `itcast_ods`.`itcast_order_refunds`(
      id                bigint,
      orderId           bigint,
      goodsId           bigint,
      refundTo          bigint,
      refundReson       bigint,
      refundOtherReson  string,
      backMoney         double,
      refundTradeNo     string,
      refundRemark      string,
      refundTime        string,
      shopRejectReason  string,
      refundStatus      bigint,
      createTime        string,
      modifiedTime        string
  )
  partitioned by (dt string)
  STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
  
  -- 创建ods层地址表
  drop table if exists `itcast_ods`.`itcast_user_address`;
  create EXTERNAL table `itcast_ods`.`itcast_user_address`(
      addressId    bigint,
      userId       bigint,
      userName     string,
      otherName    string,
      userPhone    string,
      areaIdPath   string,
      areaId       bigint,
      userAddress  string,
      isDefault    bigint,
      dataFlag     bigint,
      createTime   string
  )
  partitioned by (dt string)
  STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
  
  -- 创建ods层支付方式表
  drop table if exists `itcast_ods`.`itcast_payments`;
  create EXTERNAL table `itcast_ods`.`itcast_payments`(
      id         bigint,
      payCode    string,
      payName    string,
      payDesc    string,
      payOrder   bigint,
      payConfig  string,
      enabled    bigint,
      isOnline   bigint,
      payFor     string
  )
  partitioned by (dt string)
  STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40
  41
  42
  43
  44
  45
  46
  47
  48
  49
  50
  51
  52
  53
  54
  55
  56
  57
  58
  59
  60
  61
  62
  63
  64
  65
  66
  67
  68
  69
  70
  71
  72
  73
  74
  75
  76
  77
  78
  79
  80
  81
  82
  83
  84
  85
  86
  87
  88
  89
  90
  91
  92
  93
  94
  95
  96
  97
  98
  99
  100
  101
  102
  103
  104
  105
  106
  107
  108
  109
  110
  111
  112
  113
  114
  115
  116
  117
  118
  119
  120
  121
  122
  123
  124
  125
  126
  127
  128
  129
  130
  131
  132
  133
  134
  135
  136
  137
  138
  139
  140
  141
  142
  143
  144
  145
  146
  147
  148
  149
  150
  151
  152
  153
  154
  155
  156
  157
  158
  159
  160
  161
  162
  163
  164
  165
  166
  167
  168
  169
  170
  171
  172
  173
  174
  175
  176
  177
  178
  179
  180
  181
  182
  183
  184
  185
  186
  187
  188
  189
  190
  191
  192
  193
  194
  195
  196
  197
  198
  199
  200
  201
  202
  203
  204
  205
  206
  207
  208
  209
  210
  211
  212
  213
  214
  215
  216
  217
  218
  219
  220
  221
  222
  223
  224
  225
  226
  227
  228
  229
  230
  231
  232
  233
  234
  235
  236
  237
  238
  239
  240
  241
  242
  243
  244
  245
  246
  247
  248
  249
  250
  251
  252
  253
  254
  255
  256
  257
  258
  259
  260
  261
  262
  263
  264
  265
  266
  267
  268
  269
  270
  271
  272
  273
  274
  275
  276
  277
  278
  279
  280
  281
  282
  283
  284
  285
  286
  287
  288
  289
  290
  291
  292
  293
  294
  295
  296
  297
  298
  299
  300
  301
  302
  303
  304
  305
  306
  307
  308
  309
  310
  311
  312
  313
  314

4、数据采集

4.1 ods层全量数据抽取

步骤一：拖拽组件构建Kettle作业结构图

步骤二：转换结构图–》配置命名参数

步骤三：配置Hive SQL脚本

msck repair table itcast_ods.itcast_orders;
msck repair table itcast_ods.itcast_goods;
msck repair table itcast_ods.itcast_order_goods;
msck repair table itcast_ods.itcast_shops;
msck repair table itcast_ods.itcast_goods_cats;
msck repair table itcast_ods.itcast_org;
msck repair table itcast_ods.itcast_order_refunds;
msck repair table itcast_ods.itcast_users;
msck repair table itcast_ods.itcast_user_address;
msck repair table itcast_ods.itcast_payments;
1
2
3
4
5
6
7
8
9
10

步骤四：配置表输入

SELECT
*
FROM itcast_orders
WHERE DATE_FORMAT(createtime, '%Y%m%d') <= '${dt}';
1
2
3
4

步骤五：配置字段选择指定日期格式，配置parquet格式并设置snappy压缩输出

测试

测试数据是否都正确被加载。

select * from itcast_ods.itcast_orders limit 2;
select * from itcast_ods.itcast_goods limit 2;
select * from itcast_ods.itcast_order_goods limit 2;
select * from itcast_ods.itcast_shops limit 2;
select * from itcast_ods.itcast_goods_cats limit 2;
select * from itcast_ods.itcast_org limit 2;
select * from itcast_ods.itcast_order_refunds limit 2;
select * from itcast_ods.itcast_users limit 2;
select * from itcast_ods.itcast_user_address limit 2;
select * from itcast_ods.itcast_payments limit 2;
1
2
3
4
5
6
7
8
9
10

注意：

• 其中itcast_orders,itcast_order_goods,itcast_order_refunds表是根据时间抽取，其余表进行全量抽取！！
• 注意使用字段选择组件时要注意修改日期格式为UTF8！！，parquet中fields中date类型改为UTF8类型！！

4.2 商品维度数据加载（使用拉链表解决SCD问题）

4.2.1 dw层建表

-- dw层建表
DROP TABLE IF EXISTS `itcast_dw`.`dim_goods`;
CREATE TABLE `itcast_dw`.`dim_goods`(
    goodsId bigint,
    goodsSn string,
    productNo string,
    goodsName string,
    goodsImg string,
    shopId bigint,
    goodsType bigint,
    marketPrice double,
    shopPrice double,
    warnStock bigint,
    goodsStock bigint,
    goodsUnit string,
    goodsTips string,
    isSale bigint,
    isBest bigint,
    isHot bigint,
    isNew bigint,
    isRecom bigint,
    goodsCatIdPath string,
    goodsCatId bigint,
    shopCatId1 bigint,
    shopCatId2 bigint,
    brandId bigint,
    goodsDesc string,
    goodsStatus bigint,
    saleNum bigint,
    saleTime string,
    visitNum bigint,
    appraiseNum bigint,
    isSpec bigint,
    gallery string,
    goodsSeoKeywords string,
    illegalRemarks string,
    dataFlag bigint,
    createTime string,
    isFreeShipping bigint,
    goodsSerachKeywords string,
    modifyTime string,
    dw_start_date string,
    dw_end_date string
)
STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

4.2.2 “拉链表”方案导入数据

1 第一次全量导入

• 所有的ODS数据全部导入到拉链历史记录表中

2 增量导入（某天，举例：2019-09-10）

• 增量导入某天的数据到ODS分区
• 合并历史数据
• 通过连接查询方式更新

4.2.2.1 全量导入

• 将所有 2019年09月09日以前创建的商品以及修改的数据全部导入到拉链历史记录表中

操作步骤：

1. 使用Kettle将20190909以前的数据抽取到ods

   SELECT *
   FROM itcast_ods.itcast_goods
   WHERE  DATE_FORMAT(modifyTime, '%Y%m%d') <= '20190909';
   1
   2
   3

2. 使用spark sql将全量数据导入到dw层维度表

   set spark.sql.shuffle.partitions=1; --shuffle时的分区数，默认是200个
   -- 使用spark sql将全量数据导入到dw层维度表
   insert overwrite table `itcast_dw`.`dim_goods`
   select
       goodsId,
       goodsSn,
       productNo,
       goodsName,
       goodsImg,
       shopId,
       goodsType,
       marketPrice,
       shopPrice,
       warnStock,
       goodsStock,
       goodsUnit,
       goodsTips,
       isSale,
       isBest,
       isHot,
       isNew,
       isRecom,
       goodsCatIdPath,
       goodsCatId,
       shopCatId1,
       shopCatId2,
       brandId,
       goodsDesc,
       goodsStatus,
       saleNum,
       saleTime,
       visitNum,
       appraiseNum,
       isSpec,
       gallery,
       goodsSeoKeywords,
       illegalRemarks,
       dataFlag,
       createTime,
       isFreeShipping,
       goodsSerachKeywords,
       modifyTime,
        case when modifyTime is not null
           then from_unixtime(unix_timestamp(modifyTime, 'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd')
           else from_unixtime(unix_timestamp(createTime, 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd') 
           end as dw_start_date,
       '9999-12-31' as dw_end_date
   from
       `itcast_ods`.`itcast_goods` t
   where dt='20190909';
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50

4.2.2.2 增量导入

• 将2019年09月10日创建的 修改的数据全部导入到历史拉链表中

操作步骤：

1. 使用Kettle将20190910创建的 或者修改的数据抽取到ods

   SELECT *
   FROM itcast_goods
   WHERE  DATE_FORMAT(modifyTime, '%Y%m%d') = '${dt}';
   1
   2
   3

2. 编写spark-sql更新历史数据

   -- 更新历史数据
   select
       dw.goodsId,
       dw.goodsSn,
       dw.productNo,
       dw.goodsName,
       dw.goodsImg,
       dw.shopId,
       dw.goodsType,
       dw.marketPrice,
       dw.shopPrice,
       dw.warnStock,
       dw.goodsStock,
       dw.goodsUnit,
       dw.goodsTips,
       dw.isSale,
       dw.isBest,
       dw.isHot,
       dw.isNew,
       dw.isRecom,
       dw.goodsCatIdPath,
       dw.goodsCatId,
       dw.shopCatId1,
       dw.shopCatId2,
       dw.brandId,
       dw.goodsDesc,
       dw.goodsStatus,
       dw.saleNum,
       dw.saleTime,
       dw.visitNum,
       dw.appraiseNum,
       dw.isSpec,
       dw.gallery,
       dw.goodsSeoKeywords,
       dw.illegalRemarks,
       dw.dataFlag,
       dw.createTime,
       dw.isFreeShipping,
       dw.goodsSerachKeywords,
       dw.modifyTime,
       dw.dw_start_date,
       case when dw.dw_end_date = '9999-12-31' and ods.goodsId is not null
           then '2019-09-09'
           else dw.dw_end_date
           end as dw_end_date
   from
       `itcast_dw`.`dim_goods` dw
       left join 
       (select * from `itcast_ods`.`itcast_goods` where dt='20190910') ods
       on dw.goodsId = ods.goodsId ;
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50

3. 编写spark-sql获取当日数据

   -- 今日数据
   select
       goodsId,
       goodsSn,
       productNo,
       goodsName,
       goodsImg,
       shopId,
       goodsType,
       marketPrice,
       shopPrice,
       warnStock,
       goodsStock,
       goodsUnit,
       goodsTips,
       isSale,
       isBest,
       isHot,
       isNew,
       isRecom,
       goodsCatIdPath,
       goodsCatId,
       shopCatId1,
       shopCatId2,
       brandId,
       goodsDesc,
       goodsStatus,
       saleNum,
       saleTime,
       visitNum,
       appraiseNum,
       isSpec,
       gallery,
       goodsSeoKeywords,
       illegalRemarks,
       dataFlag,
       createTime,
       isFreeShipping,
       goodsSerachKeywords,
       modifyTime,
       case when modifyTime is not null
           then from_unixtime(unix_timestamp(modifyTime, 'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd')
           else from_unixtime(unix_timestamp(createTime, 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd') 
           end as dw_start_date,
       '9999-12-31' as dw_end_date
   from
       `itcast_ods`.`itcast_goods`
   where dt = '20190910';
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48

4. 将历史数据 当日数据合并加载到临时表

   -- 将历史数据 当日数据合并加载到临时表
   drop table if exists `itcast_dw`.`tmp_dim_goods_history`;
   create table `itcast_dw`.`tmp_dim_goods_history`
   as
   select
       dw.goodsId,
       dw.goodsSn,
       dw.productNo,
       dw.goodsName,
       dw.goodsImg,
       dw.shopId,
       dw.goodsType,
       dw.marketPrice,
       dw.shopPrice,
       dw.warnStock,
       dw.goodsStock,
       dw.goodsUnit,
       dw.goodsTips,
       dw.isSale,
       dw.isBest,
       dw.isHot,
       dw.isNew,
       dw.isRecom,
       dw.goodsCatIdPath,
       dw.goodsCatId,
       dw.shopCatId1,
       dw.shopCatId2,
       dw.brandId,
       dw.goodsDesc,
       dw.goodsStatus,
       dw.saleNum,
       dw.saleTime,
       dw.visitNum,
       dw.appraiseNum,
       dw.isSpec,
       dw.gallery,
       dw.goodsSeoKeywords,
       dw.illegalRemarks,
       dw.dataFlag,
       dw.createTime,
       dw.isFreeShipping,
       dw.goodsSerachKeywords,
       dw.modifyTime,
       dw.dw_start_date,
       case when dw.dw_end_date >= '9999-12-31' and ods.goodsId is not null
           then '2019-09-09'
           else dw.dw_end_date
           end as dw_end_date
   from
       `itcast_dw`.`dim_goods` dw
       left join 
       (select * from `itcast_ods`.`itcast_goods` where dt='20190910') ods
       on dw.goodsId = ods.goodsId
   union all
   select
       goodsId,
       goodsSn,
       productNo,
       goodsName,
       goodsImg,
       shopId,
       goodsType,
       marketPrice,
       shopPrice,
       warnStock,
       goodsStock,
       goodsUnit,
       goodsTips,
       isSale,
       isBest,
       isHot,
       isNew,
       isRecom,
       goodsCatIdPath,
       goodsCatId,
       shopCatId1,
       shopCatId2,
       brandId,
       goodsDesc,
       goodsStatus,
       saleNum,
       saleTime,
       visitNum,
       appraiseNum,
       isSpec,
       gallery,
       goodsSeoKeywords,
       illegalRemarks,
       dataFlag,
       createTime,
       isFreeShipping,
       goodsSerachKeywords,
       modifyTime,
       case when modifyTime is not null
           then from_unixtime(unix_timestamp(modifyTime, 'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd')
           else from_unixtime(unix_timestamp(createTime, 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd') 
           end as dw_start_date,
       '9999-12-31' as dw_end_date
   from
       `itcast_ods`.`itcast_goods`
   where dt = '20190910';
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50
   51
   52
   53
   54
   55
   56
   57
   58
   59
   60
   61
   62
   63
   64
   65
   66
   67
   68
   69
   70
   71
   72
   73
   74
   75
   76
   77
   78
   79
   80
   81
   82
   83
   84
   85
   86
   87
   88
   89
   90
   91
   92
   93
   94
   95
   96
   97
   98
   99
   100
   101

5. 将历史数据 当日数据导入到历史拉链表

   -- 将历史数据 当日数据导入到历史拉链表
   insert overwrite table `itcast_dw`.`dim_goods`
   select * from `itcast_dw`.`tmp_dim_goods_history`;
   
   -- 获取2019-09-09日的商品数据
   select * from `itcast_dw`.`dim_goods` where dw_start_date <= '2019-09-09' and dw_end_date >= '2019-09-09' limit 10;
   --查看对应商品id的历史拉链数据
   select * from `itcast_dw`.`dim_goods` where goodsId = 100134;
   1
   2
   3
   4
   5
   6
   7
   8

4.3 周期性事实表（使用拉链表解决SCD问题）

因为订单表和订单退款表都有状态变化的特点，所以他们作为周期性事实表在进行同步操作也就是采集数据到数仓中时需要我们能记录下订单的状态变化。因次依然使用拉链表来解决这类周期性事实表的同步需求。订单明细表并不会随着时间而变化，所以不需要使用拉链表进行同步。

订单表、订单退款表 拉链表具体实现步骤：

4.3.1 创建dw层订单拉链表、订单退款拉链表

-- 创建dw层订单事实表--带有分区字段
DROP TABLE IF EXISTS itcast_dw.fact_orders;
create  table itcast_dw.fact_orders(
    orderId            bigint,
    orderNo            string,
    shopId             bigint,
    userId             bigint,
    orderStatus        bigint,
    goodsMoney         double,
    deliverType        bigint,
    deliverMoney       double,
    totalMoney         double,
    realTotalMoney     double,
    payType            bigint,
    isPay              bigint,
    areaId             bigint,
    userAddressId      bigint,
    areaIdPath         string,
    userName           string,
    userAddress        string,
    userPhone          string,
    orderScore         bigint,
    isInvoice          bigint,
    invoiceClient      string,
    orderRemarks       string,
    orderSrc           bigint,
    needPay            double,
    payRand            bigint,
    orderType          bigint,
    isRefund           bigint,
    isAppraise         bigint,
    cancelReason       bigint,
    rejectReason       bigint,
    rejectOtherReason  string,
    isClosed           bigint,
    goodsSearchKeys    string,
    orderunique        string,
    receiveTime        string,
    deliveryTime       string,
    tradeNo            string,
    dataFlag           bigint,
    createTime         string,
    settlementId       bigint,
    commissionFee      double,
    scoreMoney         double,
    useScore           bigint,
    orderCode          string,
    extraJson          string,
    orderCodeTargetId  bigint,
    noticeDeliver      bigint,
    invoiceJson        string,
    lockCashMoney      double,
    payTime            string,
    isBatch            bigint,
    totalPayFee        bigint,
    modifiedTime        string,
    dw_start_date       string,
    dw_end_date         string
)
partitioned by (dt string) --按照天分区
STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
--临时订单表
DROP TABLE IF EXISTS itcast_dw.tmp_fact_orders;
create  table itcast_dw.tmp_fact_orders(
    orderId            bigint,
    orderNo            string,
    shopId             bigint,
    userId             bigint,
    orderStatus        bigint,
    goodsMoney         double,
    deliverType        bigint,
    deliverMoney       double,
    totalMoney         double,
    realTotalMoney     double,
    payType            bigint,
    isPay              bigint,
    areaId             bigint,
    userAddressId      bigint,
    areaIdPath         string,
    userName           string,
    userAddress        string,
    userPhone          string,
    orderScore         bigint,
    isInvoice          bigint,
    invoiceClient      string,
    orderRemarks       string,
    orderSrc           bigint,
    needPay            double,
    payRand            bigint,
    orderType          bigint,
    isRefund           bigint,
    isAppraise         bigint,
    cancelReason       bigint,
    rejectReason       bigint,
    rejectOtherReason  string,
    isClosed           bigint,
    goodsSearchKeys    string,
    orderunique        string,
    receiveTime        string,
    deliveryTime       string,
    tradeNo            string,
    dataFlag           bigint,
    createTime         string,
    settlementId       bigint,
    commissionFee      double,
    scoreMoney         double,
    useScore           bigint,
    orderCode          string,
    extraJson          string,
    orderCodeTargetId  bigint,
    noticeDeliver      bigint,
    invoiceJson        string,
    lockCashMoney      double,
    payTime            string,
    isBatch            bigint,
    totalPayFee        bigint,
    modifiedTime        string,
    dw_start_date       string,
    dw_end_date         string
)
partitioned by (dt string)
STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');

--创建订单退款表--带有分区字段
drop table if exists `itcast_dw`.`fact_order_refunds`;
create  table `itcast_dw`.`fact_order_refunds`(
    id                bigint,
    orderId           bigint,
    goodsId           bigint,
    refundTo          bigint,
    refundReson       bigint,
    refundOtherReson  string,
    backMoney         double,
    refundTradeNo     string,
    refundRemark      string,
    refundTime        string,
    shopRejectReason  string,
    refundStatus      bigint,
    createTime        string,
    modifiedTime        string,
    dw_start_date string,
  dw_end_date string
)
partitioned by (dt string) --按照天分区
STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
--临时表
drop table if exists `itcast_dw`.`tmp_fact_order_refunds`;
create  table `itcast_dw`.`tmp_fact_order_refunds`(
    id                bigint,
    orderId           bigint,
    goodsId           bigint,
    refundTo          bigint,
    refundReson       bigint,
    refundOtherReson  string,
    backMoney         double,
    refundTradeNo     string,
    refundRemark      string,
    refundTime        string,
    shopRejectReason  string,
    refundStatus      bigint,
    createTime        string,
    modifiedTime        string,
    dw_start_date string,
  dw_end_date string
)
partitioned by (dt string)
STORED AS PARQUET TBLPROPERTIES('parquet.compression'='SNAPPY');
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167

4.3.2 第一次全量数据导入拉链表

我们开启hive的动态分区，并根据数据的createtime字段进行分区划分，同一天创建的订单放在同一分区！！

#开启动态分区，默认是false
#开启允许所有分区都是动态的，否则必须要有静态分区才能使用
set hive.exec.dynamici.partition=true;  
set hive.exec.dynamic.partition.mode=nonstrict; 

1
2
3
4
5

订单表数据：ods层导入dw层

insert overwrite table itcast_dw.fact_orders 
select
orderId            ,
orderNo            ,
shopId             ,
userId             ,
orderStatus        ,
goodsMoney         ,
deliverType        ,
deliverMoney       ,
totalMoney         ,
realTotalMoney     ,
payType            ,
isPay              ,
areaId             ,
userAddressId      ,
areaIdPath         ,
userName           ,
userAddress        ,
userPhone          ,
orderScore         ,
isInvoice          ,
invoiceClient      ,
orderRemarks       ,
orderSrc           ,
needPay            ,
payRand            ,
orderType          ,
isRefund           ,
isAppraise         ,
cancelReason       ,
rejectReason       ,
rejectOtherReason  ,
isClosed           ,
goodsSearchKeys    ,
orderunique        ,
receiveTime        ,
deliveryTime       ,
tradeNo            ,
dataFlag           ,
createTime         ,
settlementId       ,
commissionFee      ,
scoreMoney         ,
useScore           ,
orderCode          ,
extraJson          ,
orderCodeTargetId  ,
noticeDeliver      ,
invoiceJson        ,
lockCashMoney      ,
payTime            ,
isBatch            ,
totalPayFee        ,
modifiedTime       ,
--增加开始时间
date_format(modifiedTime,'yyyy-MM-dd') as dw_start_date,
--增加结束时间
'9999-12-31' as dw_end_date,
--指定动态分区使用的字段，动态分区的用法：就是查询字段的最后一个字段hive表进行解析然后存入指定分区
--此次数据分区按照订单的创建时间
date_format(createtime,'yyyyMMdd') 
from itcast_ods.itcast_orders where dt="20190909";

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

订单退款表：ods层导入dw层

insert overwrite table itcast_dw.fact_order_refunds 
select
id,
orderId,
goodsId,
refundTo,
refundReson,
refundOtherReson,
backMoney,
refundTradeNo,
refundRemark,
refundTime,
shopRejectReason,
refundStatus,
createTime,
modifiedTime,
date_format(modifiedTime,'yyyy-MM-dd') as dw_start_date,
'9999-12-31' as dw_end_date,
--此次数据分区按照订单退款的创建时间
date_format(createTime,'yyyyMMdd')
from itcast_ods.itcast_order_refunds where dt="20190909"; 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

4.3.3 增量数据导入拉链表

4.3.3.1 kettle抽取增量数据导入ods层

表输入组件sql语句：

SELECT *
FROM itcast_orders

WHERE DATE_FORMAT(modifiedTime, '%Y%m%d') = '${dt}';
1
2
3
4

4.3.3.2 ods层数据合并到dw层拉链表中

insert overwrite table itcast_dw.tmp_fact_orders  
select
dw.orderId            ,
dw.orderNo            ,
dw.shopId             ,
dw.userId             ,
dw.orderStatus        ,
dw.goodsMoney         ,
dw.deliverType        ,
dw.deliverMoney       ,
dw.totalMoney         ,
dw.realTotalMoney     ,
dw.payType            ,
dw.isPay              ,
dw.areaId             ,
dw.userAddressId      ,
dw.areaIdPath         ,
dw.userName           ,
dw.userAddress        ,
dw.userPhone          ,
dw.orderScore         ,
dw.isInvoice          ,
dw.invoiceClient      ,
dw.orderRemarks       ,
dw.orderSrc           ,
dw.needPay            ,
dw.payRand            ,
dw.orderType          ,
dw.isRefund           ,
dw.isAppraise         ,
dw.cancelReason       ,
dw.rejectReason       ,
dw.rejectOtherReason  ,
dw.isClosed           ,
dw.goodsSearchKeys    ,
dw.orderunique        ,
dw.receiveTime        ,
dw.deliveryTime       ,
dw.tradeNo            ,
dw.dataFlag           ,
dw.createTime         ,
dw.settlementId       ,
dw.commissionFee      ,
dw.scoreMoney         ,
dw.useScore           ,
dw.orderCode          ,
dw.extraJson          ,
dw.orderCodeTargetId  ,
dw.noticeDeliver      ,
dw.invoiceJson        ,
dw.lockCashMoney      ,
dw.payTime            ,
dw.isBatch            ,
dw.totalPayFee        ,
dw.modifiedTime ,
dw.dw_start_date,
--修改end_date
case when ods.orderid is not null and dw.dw_end_date ='9999-12-31'
then '2019-09-09'
else dw.dw_end_date
end as dw_end_date,
--动态分区需要的字段
dw.dt
from 
itcast_dw.fact_orders  dw 
left join 
(select * from itcast_ods.itcast_orders where dt ='20190910') ods
on dw.orderid=ods.orderid 
union all
--今天新增数据的插入动作
select
orderId            ,
orderNo            ,
shopId             ,
userId             ,
orderStatus        ,
goodsMoney         ,
deliverType        ,
deliverMoney       ,
totalMoney         ,
realTotalMoney     ,
payType            ,
isPay              ,
areaId             ,
userAddressId      ,
areaIdPath         ,
userName           ,
userAddress        ,
userPhone          ,
orderScore         ,
isInvoice          ,
invoiceClient      ,
orderRemarks       ,
orderSrc           ,
needPay            ,
payRand            ,
orderType          ,
isRefund           ,
isAppraise         ,
cancelReason       ,
rejectReason       ,
rejectOtherReason  ,
isClosed           ,
goodsSearchKeys    ,
orderunique        ,
receiveTime        ,
deliveryTime       ,
tradeNo            ,
dataFlag           ,
createTime         ,
settlementId       ,
commissionFee      ,
scoreMoney         ,
useScore           ,
orderCode          ,
extraJson          ,
orderCodeTargetId  ,
noticeDeliver      ,
invoiceJson        ,
lockCashMoney      ,
payTime            ,
isBatch            ,
totalPayFee        ,
modifiedTime       ,
--增加开始时间
date_format(modifiedTime,'yyyy-MM-dd') as dw_start_date,
--增加结束时间
'9999-12-31' as dw_end_date,
--指定动态分区使用的字段，动态分区的用法：就是查询字段的最后一个字段hive表进行解析然后存入指定分区
--此次数据分区按照订单的创建时间
date_format(createtime,'yyyyMMdd') 
from itcast_ods.itcast_orders where dt="20190910";
--从临时表再插入itcast_dw.fact_orders
insert overwrite table itcast_dw.fact_orders
select
* from
itcast_dw.tmp_fact_orders;
--验证数据查询拉链表数据
select * from itcast_dw.fact_orders limit 5;


--订单退款表增量数据与历史数据合并覆盖插入dw层临时拉链表中
insert overwrite table itcast_dw.tmp_fact_order_refunds
select
dw.id,
dw.orderId,
dw.goodsId,
dw.refundTo,
dw.refundReson,
dw.refundOtherReson,
dw.backMoney,
dw.refundTradeNo,
dw.refundRemark,
dw.refundTime,
dw.shopRejectReason,
dw.refundStatus,
dw.createTime,
dw.modifiedTime,
dw.dw_start_date,
case when ods.id is not null and dw.dw_end_date ='9999-12-31'
then '2019-09-09'
else dw.dw_end_date
end as dw_end_date,
dw.dt
from  itcast_dw.fact_order_refunds  dw
left join (select * from itcast_ods.itcast_order_refunds where dt="20190910") ods
on dw.id =ods.id 
union all 
select
id,
orderId,
goodsId,
refundTo,
refundReson,
refundOtherReson,
backMoney,
refundTradeNo,
refundRemark,
refundTime,
shopRejectReason,
refundStatus,
createTime,
modifiedTime,
date_format(modifiedTime,'yyyy-MM-dd') as dw_start_date,
'9999-12-31' as dw_end_date,
date_format(createTime,'yyyyMMdd')
from itcast_ods.itcast_order_refunds where dt="20190910";
--合并数据插入临时表
insert overwrite table itcast_dw.fact_order_refunds
select * from itcast_dw.tmp_fact_order_refunds;
--验证数据
select * from itcast_dw.fact_order_refunds limit 5;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192

4.3.3.3 拉链表分区意义

--合并11号数据
insert overwrite table itcast_dw.tmp_fact_orders  
select
dw.orderId            ,
dw.orderNo            ,
dw.shopId             ,
dw.userId             ,
dw.orderStatus        ,
dw.goodsMoney         ,
dw.deliverType        ,
dw.deliverMoney       ,
dw.totalMoney         ,
dw.realTotalMoney     ,
dw.payType            ,
dw.isPay              ,
dw.areaId             ,
dw.userAddressId      ,
dw.areaIdPath         ,
dw.userName           ,
dw.userAddress        ,
dw.userPhone          ,
dw.orderScore         ,
dw.isInvoice          ,
dw.invoiceClient      ,
dw.orderRemarks       ,
dw.orderSrc           ,
dw.needPay            ,
dw.payRand            ,
dw.orderType          ,
dw.isRefund           ,
dw.isAppraise         ,
dw.cancelReason       ,
dw.rejectReason       ,
dw.rejectOtherReason  ,
dw.isClosed           ,
dw.goodsSearchKeys    ,
dw.orderunique        ,
dw.receiveTime        ,
dw.deliveryTime       ,
dw.tradeNo            ,
dw.dataFlag           ,
dw.createTime         ,
dw.settlementId       ,
dw.commissionFee      ,
dw.scoreMoney         ,
dw.useScore           ,
dw.orderCode          ,
dw.extraJson          ,
dw.orderCodeTargetId  ,
dw.noticeDeliver      ,
dw.invoiceJson        ,
dw.lockCashMoney      ,
dw.payTime            ,
dw.isBatch            ,
dw.totalPayFee        ,
dw.modifiedTime ,
dw.dw_start_date,
--修改end_date
case when ods.orderid is not null and dw.dw_end_date ='9999-12-31'
then '2019-09-10'
else dw.dw_end_date
end
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61