devbox
Li_阴宅
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink 实时数仓（十）【DWS 层搭建（四）交易域汇总表创建】

作者：Li_阴宅 | 2024-08-13 00:16:54

赞

踩

Flink 实时数仓（十）【DWS 层搭建（四）交易域汇总表创建】

前言

        今天完成 DWS 层交易域剩下的两个指标，估计一早上就完了，这两个需求用到的知识点和昨天的需求差不多；

1、交易域省份粒度下单各窗口汇总表

1.1、思路分析

        这个需求是比较简单的，province_id 字段是订单表中的字段，在 DWD 层的下单事务事实表中我们已经将该字段保留下来了，所以我们只需要读取 DWD 层的下单事务事实表即可；考虑到下单事务事实表来源于订单与处理表，而这张表在生成时需要和 活动、优惠券进行 left join，所以依然是迟到数据造成的数据重复问题，不过依然不影响，因为那俩表的字段我们用不上，所以我们只需要过滤出第一条数据即可完成去重；

  • 消费 dwd_trade_order_detail
  • 转为 JSON 流
  • 第一次去重（取迟到数据中的第一条即可）
  • 转为 JavaBean 流
  • 提取事件时间（取 create_time）并设置水位线
  • 按照粒度（province_id）分组
  • 开窗（依旧是 10s 的滚动窗口）
  • 聚合（依旧是增量聚合 + 全量聚合）
  • 关联省份信息补全 province_name
  • 写出到 clickhouse

1.2、代码实现

1.2.1、创建 ck 表及其实体类

字段依然是 粒度 + 窗口起止时间 + 度量值

  1. create table if not exists dws_trade_province_order_window
  2. (
  3.     stt           DateTime,
  4.     edt           DateTime,
  5.     province_id   String,
  6.     province_name String,
  7.     order_count   UInt64,
  8.     order_amount  Decimal(38, 20),
  9.     ts            UInt64
  10. ) engine = ReplacingMergeTree(ts)
  11.       partition by toYYYYMMDD(stt)
  12.       order by (stt, edt, province_id);
  1. import lombok.AllArgsConstructor;
  2. import lombok.Builder;
  3. import lombok.Data;
  4.  
  5. import java.util.Set;
  6.  
  7. @Data
  8. @AllArgsConstructor
  9. @Builder
  10. public class TradeProvinceOrderWindow {
  11.     // 窗口起始时间
  12.     String stt;
  13.  
  14.     // 窗口结束时间
  15.     String edt;
  16.  
  17.     // 省份 ID
  18.     String provinceId;
  19.  
  20.     // 省份名称
  21.     @Builder.Default
  22.     String provinceName = "";
  23.  
  24.     // 累计下单次数
  25.     Long orderCount;
  26.  
  27.     // 订单 ID 集合，用于统计下单次数
  28.     @TransientSink
  29.     Set<String> orderIdSet;
  30.  
  31.     // 累计下单金额
  32.     Double orderAmount;
  33.  
  34.     // 时间戳
  35.     Long ts;
  36. }

1.2.2、读取下单事务事实表并转为 JSON 流

  1. // TODO 2. 读取 kafka dwd_trade_order_detail
  2.         String groupId = "dws_trade_province_order_window";
  3.         DataStreamSource<String> orderDetailDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_trade_order_detail", groupId));
  4.  
  5.         // TODO 3. 转为 JSON 流
  6.         SingleOutputStreamOperator<JSONObject> jsonDS = orderDetailDS.flatMap(new RichFlatMapFunction<String, JSONObject>() {
  7.             @Override
  8.             public void flatMap(String value, Collector<JSONObject> out) throws Exception {
  9.                 try {
  10.                     JSONObject jsonObject = JSONObject.parseObject(value);
  11.                     out.collect(jsonObject);
  12.                 } catch (Exception e) {
  13.                     // 可以选择输出到侧输出流
  14.                     e.printStackTrace();
  15.                 }
  16.             }
  17.         });

1.2.3、第一次去重（上游 left join 迟到数据）

这里因为我们的数据和 left join 的右表无关，所以直接取第一条即可

  1. // TODO 4. 第一次去重(根据 order_detail_id 进行分组)
  2.         KeyedStream<JSONObject, String> keyedByIdStream = jsonDS.keyBy(json -> json.getString("id"));
  3.         SingleOutputStreamOperator<JSONObject> filterDS = keyedByIdStream.filter(new RichFilterFunction<JSONObject>() {
  4.             // 用来存储第一条数据,之后的数据来了全部丢弃(上游 left join 迟到的数据)
  5.             private ValueState<String> state;
  6.  
  7.             @Override
  8.             public void open(Configuration parameters) throws Exception {
  9.                 StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.seconds(5))
  10.                         .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
  11.                         .build();
  12.  
  13.                 ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("first-value", String.class);
  14.                 stateDescriptor.enableTimeToLive(ttlConfig);
  15.                 state = getRuntimeContext().getState(stateDescriptor);
  16.             }
  17.  
  18.             @Override
  19.             public boolean filter(JSONObject value) throws Exception {
  20.                 String data = state.value();
  21.                 if (data == null) {
  22.                     state.update("1"); //  随便存就行
  23.                     return true;
  24.                 }
  25.                 return false;
  26.             }
  27.         });

1.2.4、转为 JavaBean 流并设置水位线

转 JavaBean 流时，取 create_time 作为 ts 字段，以供下面提取它为事件时间（虽然之后还会把它设为系统时间作为 ck 表的版本字段）

  1. // TODO 5. 转为 JavaBean 流
  2.         SingleOutputStreamOperator<TradeProvinceOrderWindow> provinceOrderDS = filterDS.map(new MapFunction<JSONObject, TradeProvinceOrderWindow>() {
  3.             @Override
  4.             public TradeProvinceOrderWindow map(JSONObject value) throws Exception {
  5.                 HashSet<String> orderIds = new HashSet<>();
  6.                 orderIds.add(value.getString("order_id"));
  7.  
  8.                 return TradeProvinceOrderWindow.builder()
  9.                         .orderAmount(value.getDouble("split_total_amount"))
  10.                         .orderIdSet(orderIds)
  11.                         .ts(DateFormatUtil.toTs(value.getString("create_time"),true))
  12.                         .provinceId(value.getString("province_id"))
  13.                         .build();
  14.             }
  15.         });
  16.  
  17.         // TODO 6. 提取事件时间并设置水位线
  18.         SingleOutputStreamOperator<TradeProvinceOrderWindow> provinceOrderWithWmDS = provinceOrderDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(2))
  19.                 .withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderWindow>() {
  20.                     @Override
  21.                     public long extractTimestamp(TradeProvinceOrderWindow element, long recordTimestamp) {
  22.                         return element.getTs();
  23.                     }
  24.                 }));

1.2.5、分组开窗聚合

按照粒度分组，度量字段求和，并在窗口闭合后补充窗口起始时间和结束时间，时间戳字段设置为当前时间：

  1. //  TODO 7. 分组开窗聚合
  2.         SingleOutputStreamOperator<TradeProvinceOrderWindow> reduceDS = provinceOrderDS.keyBy(TradeProvinceOrderWindow::getProvinceId)
  3.                 .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
  4.                 .reduce(new ReduceFunction<TradeProvinceOrderWindow>() {
  5.                     @Override
  6.                     public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1, TradeProvinceOrderWindow value2) throws Exception {
  7.                         value1.getOrderIdSet().addAll(value2.getOrderIdSet());
  8.                         value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
  9.                         return value1;
  10.                     }
  11.                 }, new WindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
  12.                     @Override
  13.                     public void apply(String s, TimeWindow window, Iterable<TradeProvinceOrderWindow> input, Collector<TradeProvinceOrderWindow> out) throws Exception {
  14.                         TradeProvinceOrderWindow next = input.iterator().next();
  15.  
  16.                         next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
  17.                         next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
  18.                         next.setTs(System.currentTimeMillis());
  19.                         next.setOrderCount((long) next.getOrderIdSet().size());
  20.  
  21.                         out.collect(next);
  22.                     }
  23.                 });

1.2.6、关联维表 dim_base_province 并写出到 ck

  1. // TODO 8. 关联维保 province_info
  2.         SingleOutputStreamOperator<TradeProvinceOrderWindow> resultDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<TradeProvinceOrderWindow>("DIM_BASE_PROVINCE") {
  3.             @Override
  4.             public String getKey(TradeProvinceOrderWindow input) {
  5.                 return input.getProvinceId();
  6.             }
  7.  
  8.             @Override
  9.             public void addAttribute(TradeProvinceOrderWindow pojo, JSONObject dimInfo) {
  10.                 pojo.setProvinceName(dimInfo.getString("NAME"));
  11.             }
  12.         }, 100, TimeUnit.SECONDS);
  13.  
  14.         // TODO 9. 写出到 clickhouse
  15.         reduceDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_trade_province_order_window values (?,?,?,?,?,?)"));
  16.  
  17.         // TODO 10. 启动任务
  18.         env.execute("DwsTradeProvinceOrderWindow");

2、交易域品牌-品类-用户粒度退单各窗口汇总表

2.1、思路分析

这里是退单，DWD 层退单事务事实表的数据仅来自 ODS 中过滤出来的退单数据、字典表（字典表被关联两次，一次获取退单类型、一次获取退单原因）等

  • 从 dwd_trade_order_refund 读取数据
  • 转为 JavaBean 流（String -> JSON -> JavaBean）
  • 关联维表 sku_info 补充 tm_id、category3_id 字段
  • 提取事件时间生成水位线
  • 分组（按照粒度）开窗聚合
  • 关联维表
    • 关联base_trademark 获得 name
    • 关联 base_category3、base_category2、base_category1 获得 id 和 name
  • 写出到 clickhouse

2.2、代码实现

2.2.1、创建 ck 表及其 JavaBean

  1. create table if not exists dws_trade_trademark_category_user_refund_window
  2. (
  3.     stt            DateTime,
  4.     edt            DateTime,
  5.     trademark_id   String,
  6.     trademark_name String,
  7.     category1_id   String,
  8.     category1_name String,
  9.     category2_id   String,
  10.     category2_name String,
  11.     category3_id   String,
  12.     category3_name String,
  13.     user_id        String,
  14.     refund_count   UInt64,
  15.     ts             UInt64
  16. ) engine = ReplacingMergeTree(ts)
  17.       partition by toYYYYMMDD(stt)
  18.       order by (stt, edt, trademark_id, trademark_name, category1_id,
  19.                 category1_name, category2_id, category2_name, category3_id, category3_name, user_id);
  1. import lombok.AllArgsConstructor;
  2. import lombok.Builder;
  3. import lombok.Data;
  4.  
  5. import java.util.Set;
  6.  
  7. @Data
  8. @AllArgsConstructor
  9. @Builder
  10. public class TradeTrademarkCategoryUserRefundBean {
  11.     // 窗口起始时间
  12.     String stt;
  13.     // 窗口结束时间
  14.     String edt;
  15.     // 品牌 ID
  16.     String trademarkId;
  17.     // 品牌名称
  18.     String trademarkName;
  19.     // 一级品类 ID
  20.     String category1Id;
  21.     // 一级品类名称
  22.     String category1Name;
  23.     // 二级品类 ID
  24.     String category2Id;
  25.     // 二级品类名称
  26.     String category2Name;
  27.     // 三级品类 ID
  28.     String category3Id;
  29.     // 三级品类名称
  30.     String category3Name;
  31.  
  32.     // 订单 ID
  33.     @TransientSink
  34.     Set<String> orderIdSet;
  35.  
  36.     // sku_id
  37.     @TransientSink
  38.     String skuId;
  39.  
  40.     // 用户 ID
  41.     String userId;
  42.     // 退单次数
  43.     Long refundCount;
  44.     // 时间戳
  45.     Long ts;
  46. }

2.2.2、读取退单事务事实表并转换格式

  1. // TODO 2. 读取 kafka dwd_trade_order_detail
  2.         String groupId = "dws_trade_trademark_category_user_refund_window";
  3.         DataStreamSource<String> orderDetailDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_trade_order_refund", groupId));
  4.  
  5.         // TODO 3. 转为 JSON 流
  6.         SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> tmCateUserDS = orderDetailDS.flatMap(new RichFlatMapFunction<String, TradeTrademarkCategoryUserRefundBean>() {
  7.             @Override
  8.             public void flatMap(String value, Collector<TradeTrademarkCategoryUserRefundBean> out) throws Exception {
  9.                 try {
  10.                     JSONObject jsonObject = JSONObject.parseObject(value);
  11.  
  12.                     HashSet<String> orderIds = new HashSet<>();
  13.                     orderIds.add(jsonObject.getString("order_id"));
  14.  
  15.                     out.collect(TradeTrademarkCategoryUserRefundBean.builder()
  16.                             .userId(jsonObject.getString("user_id"))
  17.                             .ts(jsonObject.getString(DateFormatUtil.toTs("create_time",true)))
  18.                             .skuId(jsonObject.getString("sku_id"))
  19.                             .orderIdSet(orderIds)
  20.                             .build()
  21.                     );
  22.                 } catch (Exception e) {
  23.                     // 可以选择输出到侧输出流
  24.                     e.printStackTrace();
  25.                 }
  26.             }
  27.         });

2.2.3、关联维表 sku_info

关联维表 sku_info 补充 tm_id、category3_id 字段

  1. // TODO 4. 关联维表 sku_info 补充 tm_id、category3_id 字段
  2.         SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> withSkuCateUserIdDS = AsyncDataStream.unorderedWait(tmCateUserDS, new DimAsyncFunction<TradeTrademarkCategoryUserRefundBean>("DIM_SKU_INFO") {
  3.             @Override
  4.             public String getKey(TradeTrademarkCategoryUserRefundBean input) {
  5.                 return input.getSkuId();
  6.             }
  7.  
  8.             @Override
  9.             public void addAttribute(TradeTrademarkCategoryUserRefundBean pojo, JSONObject dimInfo) {
  10.                 pojo.setCategory3Id(dimInfo.getString("CATEGORY3_ID"));
  11.                 pojo.setTrademarkId(dimInfo.getString("TM_ID"));
  12.             }
  13.         }, 60 * 5, TimeUnit.SECONDS);

2.2.4、设置水位线并分组开窗聚合

  1. // TODO 5. 设置水位线
  2.         SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> withWmDS = withSkuCateUserIdDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeTrademarkCategoryUserRefundBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
  3.                 .withTimestampAssigner(new SerializableTimestampAssigner<TradeTrademarkCategoryUserRefundBean>() {
  4.                     @Override
  5.                     public long extractTimestamp(TradeTrademarkCategoryUserRefundBean element, long recordTimestamp) {
  6.                         return element.getTs();
  7.                     }
  8.                 }));
  9.  
  10.         // TODO 6. 分组（按照粒度）开窗聚合
  11.         SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> reduceDS = withWmDS.keyBy(data -> Tuple3.of(data.getCategory3Id(), data.getTrademarkId(), data.getUserId()))
  12.                 .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
  13.                 .reduce(new ReduceFunction<TradeTrademarkCategoryUserRefundBean>() {
  14.                     @Override
  15.                     public TradeTrademarkCategoryUserRefundBean reduce(TradeTrademarkCategoryUserRefundBean value1, TradeTrademarkCategoryUserRefundBean value2) throws Exception {
  16.                         value1.getOrderIdSet().addAll(value2.getOrderIdSet());
  17.                         return value1;
  18.                     }
  19.                 }, new WindowFunction<TradeTrademarkCategoryUserRefundBean, TradeTrademarkCategoryUserRefundBean, Tuple3<String, String, String>, TimeWindow>() {
  20.                     @Override
  21.                     public void apply(Tuple3<String, String, String> stringStringStringTuple3, TimeWindow window, Iterable<TradeTrademarkCategoryUserRefundBean> input, Collector<TradeTrademarkCategoryUserRefundBean> out) throws Exception {
  22.                         TradeTrademarkCategoryUserRefundBean next = input.iterator().next();
  23.                         next.setTs(System.currentTimeMillis());
  24.                         next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
  25.                         next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
  26.                         next.setRefundCount((long) next.getOrderIdSet().size());
  27.                         out.collect(next);
  28.                     }
  29.                 });

2.2.5、关联其它维表并写出数据到 ck

  1. // TODO  7. 关联其它维表
  2.         // TODO 7.2 关联 tm
  3.         SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> reduceWithTmDS = AsyncDataStream.unorderedWait(reduceDS,
  4.                 new DimAsyncFunction<TradeTrademarkCategoryUserRefundBean>("DIM_BASE_TRADEMARK") {
  5.                     @Override
  6.                     public String getKey(TradeTrademarkCategoryUserRefundBean input) {
  7.                         return input.getTrademarkId();
  8.                     }
  9.  
  10.                     @Override
  11.                     public void addAttribute(TradeTrademarkCategoryUserRefundBean pojo, JSONObject dimInfo) {
  12.                         pojo.setTrademarkName(dimInfo.getString("TM_NAME"));
  13.                     }
  14.                 },
  15.                 100, TimeUnit.SECONDS);
  16.         // TODO 7.3 关联 category3
  17.         SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> reduceWithCate3DS = AsyncDataStream.unorderedWait(reduceWithTmDS,
  18.                 new DimAsyncFunction<TradeTrademarkCategoryUserRefundBean>("DIM_BASE_CATEGORY3") {
  19.                     @Override
  20.                     public String getKey(TradeTrademarkCategoryUserRefundBean input) {
  21.                         return input.getCategory3Id();
  22.                     }
  23.  
  24.                     @Override
  25.                     public void addAttribute(TradeTrademarkCategoryUserRefundBean pojo, JSONObject dimInfo) {
  26.                         pojo.setCategory3Name(dimInfo.getString("NAME"));
  27.                         pojo.setCategory2Id("CATEGORY2_ID");
  28.                     }
  29.                 },
  30.                 100, TimeUnit.SECONDS);
  31.         // TODO 7.4 关联 category2
  32.         SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> reduceWithCate2DS = AsyncDataStream.unorderedWait(reduceWithCate3DS,
  33.                 new DimAsyncFunction<TradeTrademarkCategoryUserRefundBean>("DIM_BASE_CATEGORY2") {
  34.                     @Override
  35.                     public String getKey(TradeTrademarkCategoryUserRefundBean input) {
  36.                         return input.getCategory2Id();
  37.                     }
  38.  
  39.                     @Override
  40.                     public void addAttribute(TradeTrademarkCategoryUserRefundBean pojo, JSONObject dimInfo) {
  41.                         pojo.setCategory2Name(dimInfo.getString("NAME"));
  42.                         pojo.setCategory1Id("CATEGORY1_ID");
  43.                     }
  44.                 },
  45.                 100, TimeUnit.SECONDS);
  46.         // TODO 7.5 关联 category1
  47.         SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> reduceWithCate1DS = AsyncDataStream.unorderedWait(reduceWithCate2DS,
  48.                 new DimAsyncFunction<TradeTrademarkCategoryUserRefundBean>("DIM_BASE_CATEGORY1") {
  49.                     @Override
  50.                     public String getKey(TradeTrademarkCategoryUserRefundBean input) {
  51.                         return input.getCategory1Id();
  52.                     }
  53.  
  54.                     @Override
  55.                     public void addAttribute(TradeTrademarkCategoryUserRefundBean pojo, JSONObject dimInfo) {
  56.                         pojo.setCategory1Name(dimInfo.getString("NAME"));
  57.                     }
  58.                 },
  59.                 100, TimeUnit.SECONDS);
  60.         
  61.         // TODO 8. 写出到 clickhouse
  62.         reduceWithCate1DS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_trade_trademark_category_user_refund_window values(?,?,?,?,?,?,?,?,?,?,?,?,?)"));
  63.  
  64.         // TODO 9. 启动任务
  65.         env.execute("DwsTradeTrademarkCategoryUserRefundWindow");

总结

        至此，DWS 层搭建完毕，总的来说每一层的逻辑差不太多，难点就是对实时数据的抽象理解；接下来就是 ADS 层，今天搞定它！

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Li_阴宅/article/detail/972221
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号