基于Flink的实时数仓

忘是亡心i 2022-12-11 09:10 358阅读 0赞

转自博客园大佬：鼬手牵佐手

实时数仓主要解决传统数仓数据时效性低的问题，实时数仓通常会用在实时的OLAP分析，实时大屏展示，实时监控报警各个场景。虽然关于实时数仓架构及技术选型与传统的离线数仓会存在差异，但是关于数仓建设的基本方法论是一致的。接下来主要介绍Flink SQL从0到1搭建一个实时数仓的demo，涉及到数据采集、存储、计算、可视化整个流程。通过本文你可以了解到：

  • 实时数仓的基本架构
  • 实时数仓的数据处理流程
  • Flink1.11的SQL新特性
  • Flink1.11存在的bug
  • 完整的操作案例

案例简介

本文以电商业务为例，展示实时数仓的数据处理流程。另外，本文旨在说明实时数仓的构建流程，所以不会涉及复杂的数据计算。为了保证案例的可操作性和完整性，本文会给出详细的操作步骤。为了方便演示，本文的所有操作都是在Flink SQL Cli中完成。

架构设计

具体的架构设计如图所示：首先通过canal解析MySQL的binlog日志，将数据存储在Kafka中。然后使用Flink SQL对原始数据进行清洗关联，并将处理之后的明细宽表写入Kafka中。维表数据存储在MySQL中，通过Flink SQL对明细宽表与维表进行join，将聚合后的数据写入MySQL，最后通过FineBI进行可视化展示。

f08ddffc314c771dd6ef216c86933dd7.png

业务数据准备

  • 订单表

    复制代码

    1. CREATE TABLE `order_info` (
    2.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '编号',
    3.   `consignee` varchar(100) DEFAULT NULL COMMENT '收货人',
    4.   `consignee_tel` varchar(20) DEFAULT NULL COMMENT '收件人电话',
    5.   `total_amount` decimal(10,2) DEFAULT NULL COMMENT '总金额',
    6.   `order_status` varchar(20) DEFAULT NULL COMMENT '订单状态',
    7.   `user_id` bigint(20) DEFAULT NULL COMMENT '用户id',
    8.   `payment_way` varchar(20) DEFAULT NULL COMMENT '付款方式',
    9.   `delivery_address` varchar(1000) DEFAULT NULL COMMENT '送货地址',
    10.   `order_comment` varchar(200) DEFAULT NULL COMMENT '订单备注',
    11.   `out_trade_no` varchar(50) DEFAULT NULL COMMENT '订单交易编号（第三方支付用)',
    12.   `trade_body` varchar(200) DEFAULT NULL COMMENT '订单描述(第三方支付用)',
    13.   `create_time` datetime DEFAULT NULL COMMENT '创建时间',
    14.   `operate_time` datetime DEFAULT NULL COMMENT '操作时间',
    15.   `expire_time` datetime DEFAULT NULL COMMENT '失效时间',
    16.   `tracking_no` varchar(100) DEFAULT NULL COMMENT '物流单编号',
    17.   `parent_order_id` bigint(20) DEFAULT NULL COMMENT '父订单编号',
    18.   `img_url` varchar(200) DEFAULT NULL COMMENT '图片路径',
    19.   `province_id` int(20) DEFAULT NULL COMMENT '地区',
    20.   PRIMARY KEY (`id`)
    21. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='订单表';

    复制代码

  • 订单详情表（order_detail）

    复制代码

    1. CREATE TABLE `order_detail` (
    2.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '编号',
    3.   `order_id` bigint(20) DEFAULT NULL COMMENT '订单编号',
    4.   `sku_id` bigint(20) DEFAULT NULL COMMENT 'sku_id',
    5.   `sku_name` varchar(200) DEFAULT NULL COMMENT 'sku名称（冗余)',
    6.   `img_url` varchar(200) DEFAULT NULL COMMENT '图片名称（冗余)',
    7.   `order_price` decimal(10,2) DEFAULT NULL COMMENT '购买价格(下单时sku价格）',
    8.   `sku_num` varchar(200) DEFAULT NULL COMMENT '购买个数',
    9.   `create_time` datetime DEFAULT NULL COMMENT '创建时间',
    10.   PRIMARY KEY (`id`)
    11. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='订单详情表';

    复制代码

  • 商品表（sku_info）

    复制代码

    1. CREATE TABLE `sku_info` (
    2.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'skuid(itemID)',
    3.   `spu_id` bigint(20) DEFAULT NULL COMMENT 'spuid',
    4.   `price` decimal(10,0) DEFAULT NULL COMMENT '价格',
    5.   `sku_name` varchar(200) DEFAULT NULL COMMENT 'sku名称',
    6.   `sku_desc` varchar(2000) DEFAULT NULL COMMENT '商品规格描述',
    7.   `weight` decimal(10,2) DEFAULT NULL COMMENT '重量',
    8.   `tm_id` bigint(20) DEFAULT NULL COMMENT '品牌(冗余)',
    9.   `category3_id` bigint(20) DEFAULT NULL COMMENT '三级分类id（冗余)',
    10.   `sku_default_img` varchar(200) DEFAULT NULL COMMENT '默认显示图片(冗余)',
    11.   `create_time` datetime DEFAULT NULL COMMENT '创建时间',
    12.   PRIMARY KEY (`id`)
    13. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='商品表';

    复制代码

  • 商品一级类目表（base_category1）

    1. CREATE TABLE `base_category1` (
    2.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '编号',
    3.   `name` varchar(10) NOT NULL COMMENT '分类名称',
    4.   PRIMARY KEY (`id`)
    5. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='一级分类表';

  • 商品二级类目表（base_category2）

    复制代码

    1. CREATE TABLE `base_category2` (
    2.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '编号',
    3.   `name` varchar(200) NOT NULL COMMENT '二级分类名称',
    4.   `category1_id` bigint(20) DEFAULT NULL COMMENT '一级分类编号',
    5.   PRIMARY KEY (`id`)
    6. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='二级分类表';

    复制代码

  • 商品三级类目表（base_category3）

    复制代码

    1. CREATE TABLE `base_category3` (
    2.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '编号',
    3.   `name` varchar(200) NOT NULL COMMENT '三级分类名称',
    4.   `category2_id` bigint(20) DEFAULT NULL COMMENT '二级分类编号',
    5.   PRIMARY KEY (`id`)
    6. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='三级分类表';

    复制代码

  • 省份表（base_province）

    复制代码

    1. CREATE TABLE `base_province` (
    2.   `id` int(20) DEFAULT NULL COMMENT 'id',
    3.   `name` varchar(20) DEFAULT NULL COMMENT '省名称',
    4.   `region_id` int(20) DEFAULT NULL COMMENT '大区id',
    5.   `area_code` varchar(20) DEFAULT NULL COMMENT '行政区位码'
    6. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

    复制代码

  • 区域表（base_region）

    1. CREATE TABLE `base_region` (
    2.   `id` int(20) NOT NULL COMMENT '大区id',
    3.   `region_name` varchar(20) DEFAULT NULL COMMENT '大区名称',
    4.    PRIMARY KEY (`id`)
    5. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

数据处理流程

ods层数据同步

关于ODS层的数据同步参见我的另一篇文章基于Canal与Flink实现数据实时增量同步(一)。主要使用canal解析MySQL的binlog日志，然后将其写入到Kafka对应的topic中。由于篇幅限制，不会对具体的细节进行说明。同步之后的结果如下图所示：

357f0c94e608c0cec64da02ba0b3ab3c.png

DIM层数据准备

本案例中将维表存储在了MySQL中，实际生产中会用HBase存储维表数据。我们主要用到两张维表：区域维表和商品维表。处理过程如下：

  • 区域维表
    首先将mydw.base_province和mydw.base_region这个主题对应的数据抽取到MySQL中，主要使用Flink SQL的Kafka数据源对应的canal-json格式，注意：在执行装载之前，需要先在MySQL中创建对应的表，本文使用的MySQL数据库的名字为dim，用于存放维表数据。如下：

    复制代码

    1. -- -------------------------
    2. --   省份
    3. --   kafka Source
    4. -- ------------------------- 
    5. DROP TABLE IF EXISTS `ods_base_province`;
    6. CREATE TABLE `ods_base_province` (
    7.   `id` INT,
    8.   `name` STRING,
    9.   `region_id` INT ,
    10.   `area_code`STRING
    11. ) WITH(
    12. 'connector' = 'kafka',
    13.  'topic' = 'mydw.base_province',
    14.  'properties.bootstrap.servers' = 'kms-3:9092',
    15.  'properties.group.id' = 'testGroup',
    16.  'format' = 'canal-json' ,
    17.  'scan.startup.mode' = 'earliest-offset' 
    18. ) ; 
    20. -- -------------------------
    21. --   省份
    22. --   MySQL Sink
    23. -- ------------------------- 
    24. DROP TABLE IF EXISTS `base_province`;
    25. CREATE TABLE `base_province` (
    26.     `id` INT,
    27.     `name` STRING,
    28.     `region_id` INT ,
    29.     `area_code`STRING,
    30.     PRIMARY KEY (id) NOT ENFORCED
    31. ) WITH (
    32.     'connector' = 'jdbc',
    33.     'url' = 'jdbc:mysql://kms-1:3306/dim',
    34.     'table-name' = 'base_province', -- MySQL中的待插入数据的表
    35.     'driver' = 'com.mysql.jdbc.Driver',
    36.     'username' = 'root',
    37.     'password' = '123qwe',
    38.     'sink.buffer-flush.interval' = '1s'
    39. );
    41. -- -------------------------
    42. --   省份
    43. --   MySQL Sink Load Data
    44. -- ------------------------- 
    45. INSERT INTO base_province
    46. SELECT *
    47. FROM ods_base_province;
    49. -- -------------------------
    50. --   区域
    51. --   kafka Source
    52. -- ------------------------- 
    53. DROP TABLE IF EXISTS `ods_base_region`;
    54. CREATE TABLE `ods_base_region` (
    55.   `id` INT,
    56.   `region_name` STRING
    57. ) WITH(
    58. 'connector' = 'kafka',
    59.  'topic' = 'mydw.base_region',
    60.  'properties.bootstrap.servers' = 'kms-3:9092',
    61.  'properties.group.id' = 'testGroup',
    62.  'format' = 'canal-json' ,
    63.  'scan.startup.mode' = 'earliest-offset' 
    64. ) ; 
    66. -- -------------------------
    67. --   区域
    68. --   MySQL Sink
    69. -- ------------------------- 
    70. DROP TABLE IF EXISTS `base_region`;
    71. CREATE TABLE `base_region` (
    72.     `id` INT,
    73.     `region_name` STRING,
    74.      PRIMARY KEY (id) NOT ENFORCED
    75. ) WITH (
    76.     'connector' = 'jdbc',
    77.     'url' = 'jdbc:mysql://kms-1:3306/dim',
    78.     'table-name' = 'base_region', -- MySQL中的待插入数据的表
    79.     'driver' = 'com.mysql.jdbc.Driver',
    80.     'username' = 'root',
    81.     'password' = '123qwe',
    82.     'sink.buffer-flush.interval' = '1s'
    83. );
    85. -- -------------------------
    86. --   区域
    87. --   MySQL Sink Load Data
    88. -- ------------------------- 
    89. INSERT INTO base_region
    90. SELECT *
    91. FROM ods_base_region;

    复制代码

    经过上面的步骤，将创建维表所需要的原始数据已经存储到了MySQL中，接下来就需要在MySQL中创建维表，我们使用上面的两张表，创建一张视图：dim_province作为维表：

    复制代码

    1. -- ---------------------------------
    2. -- DIM层,区域维表,
    3. -- 在MySQL中创建视图
    4. -- ---------------------------------
    5. DROP VIEW IF EXISTS dim_province;
    6. CREATE VIEW dim_province AS
    7. SELECT
    8.   bp.id AS province_id,
    9.   bp.name AS province_name,
    10.   br.id AS region_id,
    11.   br.region_name AS region_name,
    12.   bp.area_code AS area_code
    13. FROM base_region br 
    14.      JOIN base_province bp ON br.id= bp.region_id
    15. ;

    复制代码

    这样我们所需要的维表：dim_province就创建好了，只需要在维表join时，使用Flink SQL创建JDBC的数据源，就可以使用该维表了。同理，我们使用相同的方法创建商品维表，具体如下：

    复制代码

    1. -- -------------------------
    2. --  一级类目表
    3. --   kafka Source
    4. -- ------------------------- 
    5. DROP TABLE IF EXISTS `ods_base_category1`;
    6. CREATE TABLE `ods_base_category1` (
    7.   `id` BIGINT,
    8.   `name` STRING
    9. )WITH(
    10.  'connector' = 'kafka',
    11.  'topic' = 'mydw.base_category1',
    12.  'properties.bootstrap.servers' = 'kms-3:9092',
    13.  'properties.group.id' = 'testGroup',
    14.  'format' = 'canal-json' ,
    15.  'scan.startup.mode' = 'earliest-offset' 
    16. ) ;
    18. -- -------------------------
    19. --  一级类目表
    20. --   MySQL Sink
    21. -- ------------------------- 
    22. DROP TABLE IF EXISTS `base_category1`;
    23. CREATE TABLE `base_category1` (
    24.     `id` BIGINT,
    25.     `name` STRING,
    26.      PRIMARY KEY (id) NOT ENFORCED
    27. ) WITH (
    28.     'connector' = 'jdbc',
    29.     'url' = 'jdbc:mysql://kms-1:3306/dim',
    30.     'table-name' = 'base_category1', -- MySQL中的待插入数据的表
    31.     'driver' = 'com.mysql.jdbc.Driver',
    32.     'username' = 'root',
    33.     'password' = '123qwe',
    34.     'sink.buffer-flush.interval' = '1s'
    35. );
    37. -- -------------------------
    38. --  一级类目表
    39. --   MySQL Sink Load Data
    40. -- ------------------------- 
    42. INSERT INTO base_category1
    43. SELECT *
    44. FROM ods_base_category1;
    46. -- -------------------------
    47. --  二级类目表
    48. --   kafka Source
    49. -- ------------------------- 
    50. DROP TABLE IF EXISTS `ods_base_category2`;
    51. CREATE TABLE `ods_base_category2` (
    52.   `id` BIGINT,
    53.   `name` STRING,
    54.   `category1_id` BIGINT
    55. )WITH(
    56. 'connector' = 'kafka',
    57.  'topic' = 'mydw.base_category2',
    58.  'properties.bootstrap.servers' = 'kms-3:9092',
    59.  'properties.group.id' = 'testGroup',
    60.  'format' = 'canal-json' ,
    61.  'scan.startup.mode' = 'earliest-offset' 
    62. ) ;
    64. -- -------------------------
    65. --  二级类目表
    66. --   MySQL Sink
    67. -- ------------------------- 
    68. DROP TABLE IF EXISTS `base_category2`;
    69. CREATE TABLE `base_category2` (
    70.     `id` BIGINT,
    71.     `name` STRING,
    72.     `category1_id` BIGINT,
    73.      PRIMARY KEY (id) NOT ENFORCED
    74. ) WITH (
    75.     'connector' = 'jdbc',
    76.     'url' = 'jdbc:mysql://kms-1:3306/dim',
    77.     'table-name' = 'base_category2', -- MySQL中的待插入数据的表
    78.     'driver' = 'com.mysql.jdbc.Driver',
    79.     'username' = 'root',
    80.     'password' = '123qwe',
    81.     'sink.buffer-flush.interval' = '1s'
    82. );
    84. -- -------------------------
    85. --  二级类目表
    86. --   MySQL Sink Load Data
    87. -- ------------------------- 
    88. INSERT INTO base_category2
    89. SELECT *
    90. FROM ods_base_category2;
    92. -- -------------------------
    93. -- 三级类目表
    94. --   kafka Source
    95. -- ------------------------- 
    96. DROP TABLE IF EXISTS `ods_base_category3`;
    97. CREATE TABLE `ods_base_category3` (
    98.   `id` BIGINT,
    99.   `name` STRING,
    100.   `category2_id` BIGINT
    101. )WITH(
    102. 'connector' = 'kafka',
    103.  'topic' = 'mydw.base_category3',
    104.  'properties.bootstrap.servers' = 'kms-3:9092',
    105.  'properties.group.id' = 'testGroup',
    106.  'format' = 'canal-json' ,
    107.  'scan.startup.mode' = 'earliest-offset' 
    108. ) ; 
    110. -- -------------------------
    111. --  三级类目表
    112. --   MySQL Sink
    113. -- ------------------------- 
    114. DROP TABLE IF EXISTS `base_category3`;
    115. CREATE TABLE `base_category3` (
    116.     `id` BIGINT,
    117.     `name` STRING,
    118.     `category2_id` BIGINT,
    119.     PRIMARY KEY (id) NOT ENFORCED
    120. ) WITH (
    121.     'connector' = 'jdbc',
    122.     'url' = 'jdbc:mysql://kms-1:3306/dim',
    123.     'table-name' = 'base_category3', -- MySQL中的待插入数据的表
    124.     'driver' = 'com.mysql.jdbc.Driver',
    125.     'username' = 'root',
    126.     'password' = '123qwe',
    127.     'sink.buffer-flush.interval' = '1s'
    128. );
    130. -- -------------------------
    131. --  三级类目表
    132. --   MySQL Sink Load Data
    133. -- ------------------------- 
    134. INSERT INTO base_category3
    135. SELECT *
    136. FROM ods_base_category3;
    138. -- -------------------------
    139. --   商品表
    140. --   Kafka Source
    141. -- ------------------------- 
    143. DROP TABLE IF EXISTS `ods_sku_info`;
    144. CREATE TABLE `ods_sku_info` (
    145.   `id` BIGINT,
    146.   `spu_id` BIGINT,
    147.   `price` DECIMAL(10,0),
    148.   `sku_name` STRING,
    149.   `sku_desc` STRING,
    150.   `weight` DECIMAL(10,2),
    151.   `tm_id` BIGINT,
    152.   `category3_id` BIGINT,
    153.   `sku_default_img` STRING,
    154.   `create_time` TIMESTAMP(0)
    155. ) WITH(
    156.  'connector' = 'kafka',
    157.  'topic' = 'mydw.sku_info',
    158.  'properties.bootstrap.servers' = 'kms-3:9092',
    159.  'properties.group.id' = 'testGroup',
    160.  'format' = 'canal-json' ,
    161.  'scan.startup.mode' = 'earliest-offset' 
    162. ) ; 
    164. -- -------------------------
    165. --   商品表
    166. --   MySQL Sink
    167. -- ------------------------- 
    168. DROP TABLE IF EXISTS `sku_info`;
    169. CREATE TABLE `sku_info` (
    170.   `id` BIGINT,
    171.   `spu_id` BIGINT,
    172.   `price` DECIMAL(10,0),
    173.   `sku_name` STRING,
    174.   `sku_desc` STRING,
    175.   `weight` DECIMAL(10,2),
    176.   `tm_id` BIGINT,
    177.   `category3_id` BIGINT,
    178.   `sku_default_img` STRING,
    179.   `create_time` TIMESTAMP(0),
    180.    PRIMARY KEY (tm_id) NOT ENFORCED
    181. ) WITH (
    182.     'connector' = 'jdbc',
    183.     'url' = 'jdbc:mysql://kms-1:3306/dim',
    184.     'table-name' = 'sku_info', -- MySQL中的待插入数据的表
    185.     'driver' = 'com.mysql.jdbc.Driver',
    186.     'username' = 'root',
    187.     'password' = '123qwe',
    188.     'sink.buffer-flush.interval' = '1s'
    189. );
    191. -- -------------------------
    192. --   商品
    193. --   MySQL Sink Load Data
    194. -- ------------------------- 
    195. INSERT INTO sku_info
    196. SELECT *
    197. FROM ods_sku_info;

    复制代码

    经过上面的步骤，我们可以将创建商品维表的基础数据表同步到MySQL中，同样需要提前创建好对应的数据表。接下来我们使用上面的基础表在mySQL的dim库中创建一张视图：dim_sku_info，用作后续使用的维表。

    复制代码

    1. -- ---------------------------------
    2. -- DIM层,商品维表,
    3. -- 在MySQL中创建视图
    4. -- ---------------------------------
    5. CREATE VIEW dim_sku_info AS
    6. SELECT
    7.   si.id AS id,
    8.   si.sku_name AS sku_name,
    9.   si.category3_id AS c3_id,
    10.   si.weight AS weight,
    11.   si.tm_id AS tm_id,
    12.   si.price AS price,
    13.   si.spu_id AS spu_id,
    14.   c3.name AS c3_name,
    15.   c2.id AS c2_id,
    16.   c2.name AS c2_name,
    17.   c3.id AS c1_id,
    18.   c3.name AS c1_name
    19. FROM
    20. (
    21.   sku_info si 
    22.   JOIN base_category3 c3 ON si.category3_id = c3.id
    23.   JOIN base_category2 c2 ON c3.category2_id =c2.id
    24.   JOIN base_category1 c1 ON c2.category1_id = c1.id
    25. );

    复制代码

    至此，我们所需要的维表数据已经准备好了，接下来开始处理DWD层的数据。

DWD层数据处理

经过上面的步骤，我们已经将所用的维表已经准备好了。接下来我们将对ODS的原始数据进行处理，加工成DWD层的明细宽表。具体过程如下：

　　

复制代码

  1. -- -------------------------
  2. --   订单详情
  3. --   Kafka Source
  4. -- ------------------------- 
  6. DROP TABLE IF EXISTS `ods_order_detail`;
  7. CREATE TABLE `ods_order_detail`(
  8.   `id` BIGINT,
  9.   `order_id` BIGINT,
  10.   `sku_id` BIGINT,
  11.   `sku_name` STRING,
  12.   `img_url` STRING,
  13.   `order_price` DECIMAL(10,2),
  14.   `sku_num` INT,
  15.   `create_time` TIMESTAMP(0)
  16. ) WITH(
  17.  'connector' = 'kafka',
  18.  'topic' = 'mydw.order_detail',
  19.  'properties.bootstrap.servers' = 'kms-3:9092',
  20.  'properties.group.id' = 'testGroup',
  21.  'format' = 'canal-json' ,
  22.  'scan.startup.mode' = 'earliest-offset' 
  23. ) ; 
  25. -- -------------------------
  26. --   订单信息
  27. --   Kafka Source
  28. -- -------------------------
  29. DROP TABLE IF EXISTS `ods_order_info`;
  30. CREATE TABLE `ods_order_info` (
  31.   `id` BIGINT,
  32.   `consignee` STRING,
  33.   `consignee_tel` STRING,
  34.   `total_amount` DECIMAL(10,2),
  35.   `order_status` STRING,
  36.   `user_id` BIGINT,
  37.   `payment_way` STRING,
  38.   `delivery_address` STRING,
  39.   `order_comment` STRING,
  40.   `out_trade_no` STRING,
  41.   `trade_body` STRING,
  42.   `create_time` TIMESTAMP(0) ,
  43.   `operate_time` TIMESTAMP(0) ,
  44.   `expire_time` TIMESTAMP(0) ,
  45.   `tracking_no` STRING,
  46.   `parent_order_id` BIGINT,
  47.   `img_url` STRING,
  48.   `province_id` INT
  49. ) WITH(
  50. 'connector' = 'kafka',
  51.  'topic' = 'mydw.order_info',
  52.  'properties.bootstrap.servers' = 'kms-3:9092',
  53.  'properties.group.id' = 'testGroup',
  54.  'format' = 'canal-json' ,
  55.  'scan.startup.mode' = 'earliest-offset' 
  56. ) ; 
  58. -- ---------------------------------
  59. -- DWD层,支付订单明细表dwd_paid_order_detail
  60. -- ---------------------------------
  61. DROP TABLE IF EXISTS dwd_paid_order_detail;
  62. CREATE TABLE dwd_paid_order_detail
  63. (
  64.   detail_id BIGINT,
  65.   order_id BIGINT,
  66.   user_id BIGINT,
  67.   province_id INT,
  68.   sku_id BIGINT,
  69.   sku_name STRING,
  70.   sku_num INT,
  71.   order_price DECIMAL(10,0),
  72.   create_time STRING,
  73.   pay_time STRING
  74.  ) WITH (
  75.     'connector' = 'kafka',
  76.     'topic' = 'dwd_paid_order_detail',
  77.     'scan.startup.mode' = 'earliest-offset',
  78.     'properties.bootstrap.servers' = 'kms-3:9092',
  79.     'format' = 'changelog-json'
  80. );
  81. -- ---------------------------------
  82. -- DWD层,已支付订单明细表
  83. -- 向dwd_paid_order_detail装载数据
  84. -- ---------------------------------
  85. INSERT INTO dwd_paid_order_detail
  86. SELECT
  87.   od.id,
  88.   oi.id order_id,
  89.   oi.user_id,
  90.   oi.province_id,
  91.   od.sku_id,
  92.   od.sku_name,
  93.   od.sku_num,
  94.   od.order_price,
  95.   oi.create_time,
  96.   oi.operate_time
  97. FROM
  98.     (
  99.     SELECT * 
  100.     FROM ods_order_info
  101.     WHERE order_status = '2' -- 已支付
  102.     ) oi JOIN
  103.     (
  104.     SELECT *
  105.     FROM ods_order_detail
  106.     ) od 
  107.     ON oi.id = od.order_id;

复制代码

ADS层数据

经过上面的步骤，我们创建了一张dwd_paid_order_detail明细宽表，并将该表存储在了Kafka中。接下来我们将使用这张明细宽表与维表进行JOIN，得到我们ADS应用层数据。

  • ads_province_index

首先在MySQL中创建对应的ADS目标表：ads_province_index

　　

复制代码

  1. CREATE TABLE ads.ads_province_index(
  2.   province_id INT(10),
  3.   area_code VARCHAR(100),
  4.   province_name VARCHAR(100),
  5.   region_id INT(10),
  6.   region_name VARCHAR(100),
  7.   order_amount DECIMAL(10,2),
  8.   order_count BIGINT(10),
  9.   dt VARCHAR(100),
  10.   PRIMARY KEY (province_id, dt) 
  11. ) ;

复制代码

向MySQL的ADS层目标装载数据：

复制代码

  1. -- Flink SQL Cli操作
  2. -- ---------------------------------
  3. -- 使用 DDL创建MySQL中的ADS层表
  4. -- 指标：1.每天每个省份的订单数
  5. --      2.每天每个省份的订单金额
  6. -- ---------------------------------
  7. CREATE TABLE ads_province_index(
  8.   province_id INT,
  9.   area_code STRING,
  10.   province_name STRING,
  11.   region_id INT,
  12.   region_name STRING,
  13.   order_amount DECIMAL(10,2),
  14.   order_count BIGINT,
  15.   dt STRING,
  16.   PRIMARY KEY (province_id, dt) NOT ENFORCED  
  17. ) WITH (
  18.     'connector' = 'jdbc',
  19.     'url' = 'jdbc:mysql://kms-1:3306/ads',
  20.     'table-name' = 'ads_province_index', 
  21.     'driver' = 'com.mysql.jdbc.Driver',
  22.     'username' = 'root',
  23.     'password' = '123qwe'
  24. );
  25. -- ---------------------------------
  26. -- dwd_paid_order_detail已支付订单明细宽表
  27. -- ---------------------------------
  28. CREATE TABLE dwd_paid_order_detail
  29. (
  30.   detail_id BIGINT,
  31.   order_id BIGINT,
  32.   user_id BIGINT,
  33.   province_id INT,
  34.   sku_id BIGINT,
  35.   sku_name STRING,
  36.   sku_num INT,
  37.   order_price DECIMAL(10,2),
  38.   create_time STRING,
  39.   pay_time STRING
  40.  ) WITH (
  41.     'connector' = 'kafka',
  42.     'topic' = 'dwd_paid_order_detail',
  43.     'scan.startup.mode' = 'earliest-offset',
  44.     'properties.bootstrap.servers' = 'kms-3:9092',
  45.     'format' = 'changelog-json'
  46. );
  48. -- ---------------------------------
  49. -- tmp_province_index
  50. -- 订单汇总临时表
  51. -- ---------------------------------
  52. CREATE TABLE tmp_province_index(
  53.     province_id INT,
  54.     order_count BIGINT,-- 订单数
  55.     order_amount DECIMAL(10,2), -- 订单金额
  56.     pay_date DATE
  57. )WITH (
  58.     'connector' = 'kafka',
  59.     'topic' = 'tmp_province_index',
  60.     'scan.startup.mode' = 'earliest-offset',
  61.     'properties.bootstrap.servers' = 'kms-3:9092',
  62.     'format' = 'changelog-json'
  63. );
  64. -- ---------------------------------
  65. -- tmp_province_index
  66. -- 订单汇总临时表数据装载
  67. -- ---------------------------------
  68. INSERT INTO tmp_province_index
  69. SELECT
  70.       province_id,
  71.       count(distinct order_id) order_count,-- 订单数
  72.       sum(order_price * sku_num) order_amount, -- 订单金额
  73.       TO_DATE(pay_time,'yyyy-MM-dd') pay_date
  74. FROM dwd_paid_order_detail
  75. GROUP BY province_id,TO_DATE(pay_time,'yyyy-MM-dd')
  76. ;
  77. -- ---------------------------------
  78. -- tmp_province_index_source
  79. -- 使用该临时汇总表，作为数据源
  80. -- ---------------------------------
  81. CREATE TABLE tmp_province_index_source(
  82.     province_id INT,
  83.     order_count BIGINT,-- 订单数
  84.     order_amount DECIMAL(10,2), -- 订单金额
  85.     pay_date DATE,
  86.     proctime as PROCTIME()   -- 通过计算列产生一个处理时间列
  87.  ) WITH (
  88.     'connector' = 'kafka',
  89.     'topic' = 'tmp_province_index',
  90.     'scan.startup.mode' = 'earliest-offset',
  91.     'properties.bootstrap.servers' = 'kms-3:9092',
  92.     'format' = 'changelog-json'
  93. );
  95. -- ---------------------------------
  96. -- DIM层,区域维表,
  97. -- 创建区域维表数据源
  98. -- ---------------------------------
  99. DROP TABLE IF EXISTS `dim_province`;
  100. CREATE TABLE dim_province (
  101.   province_id INT,
  102.   province_name STRING,
  103.   area_code STRING,
  104.   region_id INT,
  105.   region_name STRING ,
  106.   PRIMARY KEY (province_id) NOT ENFORCED
  107. ) WITH (
  108.     'connector' = 'jdbc',
  109.     'url' = 'jdbc:mysql://kms-1:3306/dim',
  110.     'table-name' = 'dim_province', 
  111.     'driver' = 'com.mysql.jdbc.Driver',
  112.     'username' = 'root',
  113.     'password' = '123qwe',
  114.     'scan.fetch-size' = '100'
  115. );
  117. -- ---------------------------------
  118. -- 向ads_province_index装载数据
  119. -- 维表JOIN
  120. -- ---------------------------------
  122. INSERT INTO ads_province_index
  123. SELECT
  124.   pc.province_id,
  125.   dp.area_code,
  126.   dp.province_name,
  127.   dp.region_id,
  128.   dp.region_name,
  129.   pc.order_amount,
  130.   pc.order_count,
  131.   cast(pc.pay_date as VARCHAR)
  132. FROM
  133. tmp_province_index_source pc
  134.   JOIN dim_province FOR SYSTEM_TIME AS OF pc.proctime as dp 
  135.   ON dp.province_id = pc.province_id;

复制代码

当提交任务之后：观察Flink WEB UI：

0832a3fc10fd0c1e594bd0849af17fc4.png

查看ADS层的ads_province_index表数据：

9339e133aac3bc225887c227d3bcd166.png

  • ads_sku_index

首先在MySQL中创建对应的ADS目标表：ads_sku_index

复制代码

  1. CREATE TABLE ads_sku_index
  2. (
  3.   sku_id BIGINT(10),
  4.   sku_name VARCHAR(100),
  5.   weight DOUBLE,
  6.   tm_id BIGINT(10),
  7.   price DOUBLE,
  8.   spu_id BIGINT(10),
  9.   c3_id BIGINT(10),
  10.   c3_name VARCHAR(100) ,
  11.   c2_id BIGINT(10),
  12.   c2_name VARCHAR(100),
  13.   c1_id BIGINT(10),
  14.   c1_name VARCHAR(100),
  15.   order_amount DOUBLE,
  16.   order_count BIGINT(10),
  17.   sku_count BIGINT(10),
  18.   dt varchar(100),
  19.   PRIMARY KEY (sku_id,dt)
  20. );

复制代码

向MySQL的ADS层目标装载数据：

复制代码

  1. -- ---------------------------------
  2. -- 使用 DDL创建MySQL中的ADS层表
  3. -- 指标：1.每天每个商品对应的订单个数
  4. --      2.每天每个商品对应的订单金额
  5. --      3.每天每个商品对应的数量
  6. -- ---------------------------------
  7. CREATE TABLE ads_sku_index
  8. (
  9.   sku_id BIGINT,
  10.   sku_name VARCHAR,
  11.   weight DOUBLE,
  12.   tm_id BIGINT,
  13.   price DOUBLE,
  14.   spu_id BIGINT,
  15.   c3_id BIGINT,
  16.   c3_name VARCHAR ,
  17.   c2_id BIGINT,
  18.   c2_name VARCHAR,
  19.   c1_id BIGINT,
  20.   c1_name VARCHAR,
  21.   order_amount DOUBLE,
  22.   order_count BIGINT,
  23.   sku_count BIGINT,
  24.   dt varchar,
  25.   PRIMARY KEY (sku_id,dt) NOT ENFORCED
  26. ) WITH (
  27.     'connector' = 'jdbc',
  28.     'url' = 'jdbc:mysql://kms-1:3306/ads',
  29.     'table-name' = 'ads_sku_index', 
  30.     'driver' = 'com.mysql.jdbc.Driver',
  31.     'username' = 'root',
  32.     'password' = '123qwe'
  33. );
  35. -- ---------------------------------
  36. -- dwd_paid_order_detail已支付订单明细宽表
  37. -- ---------------------------------
  38. CREATE TABLE dwd_paid_order_detail
  39. (
  40.   detail_id BIGINT,
  41.   order_id BIGINT,
  42.   user_id BIGINT,
  43.   province_id INT,
  44.   sku_id BIGINT,
  45.   sku_name STRING,
  46.   sku_num INT,
  47.   order_price DECIMAL(10,2),
  48.   create_time STRING,
  49.   pay_time STRING
  50.  ) WITH (
  51.     'connector' = 'kafka',
  52.     'topic' = 'dwd_paid_order_detail',
  53.     'scan.startup.mode' = 'earliest-offset',
  54.     'properties.bootstrap.servers' = 'kms-3:9092',
  55.     'format' = 'changelog-json'
  56. );
  58. -- ---------------------------------
  59. -- tmp_sku_index
  60. -- 商品指标统计
  61. -- ---------------------------------
  62. CREATE TABLE tmp_sku_index(
  63.     sku_id BIGINT,
  64.     order_count BIGINT,-- 订单数
  65.     order_amount DECIMAL(10,2), -- 订单金额
  66.  order_sku_num BIGINT,
  67.     pay_date DATE
  68. )WITH (
  69.     'connector' = 'kafka',
  70.     'topic' = 'tmp_sku_index',
  71.     'scan.startup.mode' = 'earliest-offset',
  72.     'properties.bootstrap.servers' = 'kms-3:9092',
  73.     'format' = 'changelog-json'
  74. );
  75. -- ---------------------------------
  76. -- tmp_sku_index
  77. -- 数据装载
  78. -- ---------------------------------
  79. INSERT INTO tmp_sku_index
  80. SELECT
  81.       sku_id,
  82.       count(distinct order_id) order_count,-- 订单数
  83.       sum(order_price * sku_num) order_amount, -- 订单金额
  84.    sum(sku_num) order_sku_num,
  85.       TO_DATE(pay_time,'yyyy-MM-dd') pay_date
  86. FROM dwd_paid_order_detail
  87. GROUP BY sku_id,TO_DATE(pay_time,'yyyy-MM-dd')
  88. ;
  90. -- ---------------------------------
  91. -- tmp_sku_index_source
  92. -- 使用该临时汇总表，作为数据源
  93. -- ---------------------------------
  94. CREATE TABLE tmp_sku_index_source(
  95.     sku_id BIGINT,
  96.     order_count BIGINT,-- 订单数
  97.     order_amount DECIMAL(10,2), -- 订单金额
  98.     order_sku_num BIGINT,
  99.     pay_date DATE,
  100.     proctime as PROCTIME()   -- 通过计算列产生一个处理时间列
  101.  ) WITH (
  102.     'connector' = 'kafka',
  103.     'topic' = 'tmp_sku_index',
  104.     'scan.startup.mode' = 'earliest-offset',
  105.     'properties.bootstrap.servers' = 'kms-3:9092',
  106.     'format' = 'changelog-json'
  107. );
  108. -- ---------------------------------
  109. -- DIM层,商品维表,
  110. -- 创建商品维表数据源
  111. -- ---------------------------------
  112. DROP TABLE IF EXISTS `dim_sku_info`;
  113. CREATE TABLE dim_sku_info (
  114.   id BIGINT,
  115.   sku_name STRING,
  116.   c3_id BIGINT,
  117.   weight DECIMAL(10,2),
  118.   tm_id BIGINT,
  119.   price DECIMAL(10,2),
  120.   spu_id BIGINT,
  121.   c3_name STRING,
  122.   c2_id BIGINT,
  123.   c2_name STRING,
  124.   c1_id BIGINT,
  125.   c1_name STRING,
  126.   PRIMARY KEY (id) NOT ENFORCED
  127. ) WITH (
  128.     'connector' = 'jdbc',
  129.     'url' = 'jdbc:mysql://kms-1:3306/dim',
  130.     'table-name' = 'dim_sku_info', 
  131.     'driver' = 'com.mysql.jdbc.Driver',
  132.     'username' = 'root',
  133.     'password' = '123qwe',
  134.     'scan.fetch-size' = '100'
  135. );
  136. -- ---------------------------------
  137. -- 向ads_sku_index装载数据
  138. -- 维表JOIN
  139. -- ---------------------------------
  140. INSERT INTO ads_sku_index
  141. SELECT
  142.   sku_id ,
  143.   sku_name ,
  144.   weight ,
  145.   tm_id ,
  146.   price ,
  147.   spu_id ,
  148.   c3_id ,
  149.   c3_name,
  150.   c2_id ,
  151.   c2_name ,
  152.   c1_id ,
  153.   c1_name ,
  154.   sc.order_amount,
  155.   sc.order_count ,
  156.   sc.order_sku_num ,
  157.   cast(sc.pay_date as VARCHAR)
  158. FROM
  159. tmp_sku_index_source sc 
  160.   JOIN dim_sku_info FOR SYSTEM_TIME AS OF sc.proctime as ds
  161.   ON ds.id = sc.sku_id
  162.   ;

复制代码

当提交任务之后：观察Flink WEB UI：

5849e327e167e99e683181ec571d549a.png

查看ADS层的ads_sku_index表数据：

94211199ad2db603803b9fdda20435b9.png

FineBI展示

7572da4dabbebf1ba45feb51b53cb7b2.png

其他注意点

Flink1.11.0存在的bug

当在代码中使用Flink1.11.0版本时，如果将一个change-log的数据源insert到一个upsert sink时，会报如下异常：

  1. [ERROR] Could not execute SQL statement. Reason:
  2. org.apache.flink.table.api.TableException: Provided trait [BEFORE_AND_AFTER] can't satisfy required trait [ONLY_UPDATE_AFTER]. This is a bug in planner, please file an issue. 
  3. Current node is TableSourceScan(table=[[default_catalog, default_database, t_pick_order]], fields=[order_no, status])

该bug目前已被修复，修复可以在Flink1.11.1中使用。

发表评论

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

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

相关阅读

    相关 基于Flink的实时数仓

    转自博客园大佬：[鼬手牵佐手][Link 1]   实时数仓主要解决传统数仓数据时效性低的问题，实时数仓通常会用在实时的OLAP分析，实时大屏展示，实时监控报警各个场景。虽

    忘是亡心i忘是亡心i/ 0 赞/ 359 阅读