主要任务

从Kafka 读取订单明细数据,过滤 null 数据并按照唯一键对数据去重,统计各省份各窗口订单数和订单金额,将数据写入ClickHouse 交易域省份粒度下单各窗口汇总表。

思路分析

1**)从 Kafka **订单明细主题读取数据

2**)转换数据结构**

3**)按照唯一键去重**

4**)转换数据结构**

  1. JSONObject 转换为实体类TradeProvinceOrderWindow

5**)设置水位线**

6**)按照省份 ID 分组**

provinceId 可以唯一标识数据。

7**)开窗**

8**)聚合计算**

度量字段求和,并在窗口闭合后补充窗口起始时间和结束时间。将时间戳置为当前系统时间。

9**)关联省份信息**

补全省份名称字段。

10)写出到ClickHouse。


图解

实时数仓(三十三)DWS层-交易域省份粒度下单各窗口汇总表 - 图1

ClickHouse 建表语句

启动ClickHouse

systemctl start clickhouse

systemctl start clickhouse-server

clickhouse-client -m

  1. drop table if exists dws_trade_province_order_window;
  2. create table if not exists dws_trade_province_order_window
  3. (
  4. stt DateTime,
  5. edt DateTime,
  6. province_id String,
  7. province_name String,
  8. order_count UInt64,
  9. order_amount Decimal(38, 20),
  10. ts UInt64
  11. ) engine = ReplacingMergeTree(ts)
  12. partition by toYYYYMMDD(stt)
  13. order by (stt, edt, province_id);

代码

(1)实体类 TradeProvinceOrderWindow

https://gitee.com/luan_hao/gmall-flink/blob/master/gmall-realtime/src/main/java/com/apache/gmall/bean/TradeProvinceOrderWindow.java

(2)主程序

https://gitee.com/luan_hao/gmall-flink/blob/master/gmall-realtime/src/main/java/com/apache/gmall/app/dws/DwsTradeProvinceOrderWindow.java

测试

开启hadoop,zookeeper,hbase,phoenix,kafka,maxwell,redis,clickhouse

启动redis 命令 :redis-cli -h hadoop102 —raw

依次启动如下三个类

DwdTradeOrderPreProcess -> Kafka(ZK) -> DwdTradeOrderDetail -> Kafka(ZK) -> DwsTradeProvinceOrderWindow(Phoenix-(HBase-HDFS、ZK)、Redis)

修改配置文件(当日)

实时数仓(三十三)DWS层-交易域省份粒度下单各窗口汇总表 - 图2

启动脚本 (首次启动后,十秒钟之后再启动一次)

实时数仓(三十三)DWS层-交易域省份粒度下单各窗口汇总表 - 图3

查看数据

实时数仓(三十三)DWS层-交易域省份粒度下单各窗口汇总表 - 图4

有如上图数据,则程序成功!