第一章.DWM层介绍

DWM层(Data WareHouse Middle),一般称为数据中间层,该层会在DWD的基础上,对数据做轻度的”聚合”操作,生成宽表

DWM层主要服务于DWS,因为部分需求直接从DWD层到DWS层中间有一定的计算量,而且这部分的计算结果很有可能被多个DWS层主题复用,所以部分DWD会形成一层DWM

主要涉及业务

  1. 访问UV计算
  2. 跳出明细计算
  3. 订单宽表
  4. 支付宽表

第二章.DWM层:UV计算

1.需求分析

UV,全称是Unique Visitor,即独立访客,对于实时计算中,也可以称为DAU( Daily Active User),即每日活跃用户,并且实时计算中的uv通常是指当日的访客数

  1.     问题:怎么从用户行为日志中识别出当日的访客
  2.     1.识别出该访客打开的第一个页面,表示这个访客开始进入我们的应用
  3.     2.由于访客可以在一天中多次进入应用,所以我们要在一天的范围内去重
  4.     3.如何在第二天对某个用户重新做UV计算

实现思路

  1. 使用event-time语义(考虑数据的乱序)
  2. 按照mid分组
  3. 添加窗口
  4. 过滤出来当天的首次访问记录(去重)
    • 使用flink的状态,而且状态只保留一天即可
    • 什么时候清除状态,现在的日期和状态中保存的日期不一致的时候清除
  5. 把当天的首次访问记录写入到dwm层(kafka)

2.具体实现

  1. package com.atguigu.gmall.app.dwm;
  3. import com.alibaba.fastjson.JSON;
  5. import com.alibaba.fastjson.JSONObject;
  6. import com.atguigu.gmall.app.BaseAppV1;
  7. import com.atguigu.gmall.common.Constant;
  8. import com.atguigu.gmall.util.FlinkSinkUtil;
  9. import com.atguigu.gmall.util.IteratorToListUtil;
  10. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  11. import org.apache.flink.api.common.functions.MapFunction;
  12. import org.apache.flink.api.common.state.ValueState;
  13. import org.apache.flink.api.common.state.ValueStateDescriptor;
  15. import org.apache.flink.configuration.Configuration;
  16. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  17. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  18. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  19. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  20. import org.apache.flink.streaming.api.windowing.time.Time;
  21. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  22. import org.apache.flink.util.Collector;
  24. import java.time.Duration;
  25. import java.util.Collections;
  26. import java.util.Comparator;
  27. import java.util.List;
  29. public class DwmUvApp extends BaseAppV1 {
  32.     public static void main(String[] args) {
  33.         new DwmUvApp().init(3001,1,"DwmUvApp","DwmUvApp", Constant.TOPIC_DWD_PAGE);
  35.     }
  38.     @Override
  39.     public void run(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
  40.         stream.map(
  41.                 new MapFunction<String, JSONObject>() {
  42.                     @Override
  43.                     public JSONObject map(String value) throws Exception {
  44.                         return JSON.parseObject(value);
  45.                     }
  46.                 }
  47.         )
  48.                 .assignTimestampsAndWatermarks(
  49.                         WatermarkStrategy
  50.                         .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  51.                         .withTimestampAssigner((obj,ts)->obj.getLong("ts"))
  52.                 )
  53.                 .keyBy(obj -> obj.getJSONObject("common").getString("mid"))
  54.                 .window(TumblingEventTimeWindows.of(Time.seconds(5)))
  55.                 .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {
  56.                     private ValueState<Long> firstVisitState;
  57.                     @Override
  58.                     public void open(Configuration parameters) throws Exception {
  59.                         firstVisitState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("firstVisitState", Long.class));
  60.                     }
  62.                     @Override
  63.                     public void process(String key, Context context, Iterable<JSONObject> elements, Collector<JSONObject> out) throws Exception {
  64.                      //识别天的变化,然后清空状态
  65.                         //今天
  66.                         String today = IteratorToListUtil.toDateString(context.window().getStart());
  67.                         //昨天
  68.                         String yesterday = firstVisitState.value() == null ? null : IteratorToListUtil.toDateString(firstVisitState.value());
  69.                         if(!today.equals(yesterday)){
  70.                             firstVisitState.clear();
  71.                         }
  72.                         if(firstVisitState.value()==null){
  73.                             List<JSONObject> list = IteratorToListUtil.toList(elements);
  74.                             JSONObject min = Collections.min(list, Comparator.comparing(ele -> ele.getLong("ts")));
  75.                             out.collect(min);
  76.                             firstVisitState.update(min.getLong("ts"));
  77.                         }
  78.                     }
  79.                 })
  80.                 .map(new MapFunction<JSONObject, String>() {
  81.                     @Override
  82.                     public String map(JSONObject value) throws Exception {
  83.                         return value.toJSONString();
  84.                     }
  85.                 })
  86.                 .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWM_UV));
  88.     }
  89. }

3.测试

分别启动hadoop,zookeeper,kafka,gmall-logger,flink-yarn,dwdLogAPP,dwmUvApp

然后在终端起一个消费者消费 dwm_uv

Flink实时数仓(二) - 图1

可以看出,消费到的mid没有重复

第三章.DWM层:跳出明细

1.需求分析

  1.     1.什么是跳出
  2.     跳出就是用户成功的访问了网站的入口页面(例如首页)后就退出,不再继续访问网站的其他页面
  3.     跳出率计算公式: 跳出率=访问一个页面后离开网站的次数 / 入口总访问次数
  4.     观察关键词的跳出率就可以得到用户对网站内容的认可,或者说你的网站是否对用户有吸引力,而网站的内容是否能够对用户有吸引力,而网站的内容是否能够对用户有帮助留住用户也可以直接从跳出率中看出来所以跳出率是衡量网站内容质量的重要标准
  5.     2.计算跳出率的思路
  6.     首先要识别哪些是跳出行为,要把这些跳出的访客最后一个访问的页面识别出来,那么要抓住几个特征:
  7.     一.该页面是用户近期访问的第一个页面,这个可以通过该页面是否有上一个页面(last_page_id)来判断,如果这个表示为空,就说明,这是这个访客访问的第一个页面
  8.     二.首次访问之后很长一段时间,用户没继续再有其他页面的访问,这个可以通过Flink自带的CEP技术来实现,用户跳出事件,本质上就是一个条件事件加一个超时事件的组合

2.具体实现

  1. package com.atguigu.gmall.app.dwm;
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.JSONObject;
  5. import com.atguigu.gmall.app.BaseAppV1;
  6. import com.atguigu.gmall.common.Constant;
  7. import com.atguigu.gmall.util.FlinkSinkUtil;
  8. import com.atguigu.gmall.util.IteratorToListUtil;
  9. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  10. import org.apache.flink.api.common.functions.MapFunction;
  11. import org.apache.flink.api.common.state.ValueState;
  12. import org.apache.flink.api.common.state.ValueStateDescriptor;
  13. import org.apache.flink.cep.CEP;
  14. import org.apache.flink.cep.PatternSelectFunction;
  15. import org.apache.flink.cep.PatternStream;
  16. import org.apache.flink.cep.PatternTimeoutFunction;
  17. import org.apache.flink.cep.pattern.Pattern;
  18. import org.apache.flink.cep.pattern.conditions.SimpleCondition;
  19. import org.apache.flink.configuration.Configuration;
  20. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  21. import org.apache.flink.streaming.api.datastream.KeyedStream;
  22. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  23. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  24. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  25. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  26. import org.apache.flink.streaming.api.windowing.time.Time;
  27. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  28. import org.apache.flink.util.Collector;
  29. import org.apache.flink.util.OutputTag;
  31. import java.time.Duration;
  32. import java.util.Collections;
  33. import java.util.Comparator;
  34. import java.util.List;
  35. import java.util.Map;
  37. public class DwmUserJumpDetailApp extends BaseAppV1 {
  40.     public static void main(String[] args) {
  41.         new DwmUserJumpDetailApp().init(3002,1,"DwmUserJumpDetailApp","DwmUserJumpDetailApp", Constant.TOPIC_DWD_PAGE);
  43.     }
  46.     @Override
  47.     public void run(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
  48.         //1.获取流
  49.         KeyedStream<JSONObject, String> keyedStream = stream.map(
  50.                 new MapFunction<String, JSONObject>() {
  51.                     @Override
  52.                     public JSONObject map(String value) throws Exception {
  53.                         return JSON.parseObject(value);
  54.                     }
  55.                 }
  56.         )
  57.                 .assignTimestampsAndWatermarks(
  58.                         WatermarkStrategy
  59.                                 .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  60.                                 .withTimestampAssigner((obj, ts) -> obj.getLong("ts"))
  61.                 )
  62.                 .keyBy(obj -> obj.getJSONObject("common").getString("mid"));
  64.         //2.定义模式
  65.         Pattern<JSONObject, JSONObject> pattern = Pattern
  66.                 .<JSONObject>begin("entry1")
  67.                 .where(new SimpleCondition<JSONObject>() {
  68.                     @Override
  69.                     public boolean filter(JSONObject value) throws Exception {
  70.                         //判断入口:lastpageId是null
  71.                         String lastPageId = value.getJSONObject("page").getString("last_page_id");
  72.                         return lastPageId == null || lastPageId.length() == 0;
  73.                     }
  74.                 })
  75.                 .next("entry2")
  76.                 .where(new SimpleCondition<JSONObject>() {
  77.                     @Override
  78.                     public boolean filter(JSONObject value) throws Exception {
  79.                         //不是入口
  80.                         String lastPageId = value.getJSONObject("page").getString("last_page_id");
  81.                         return lastPageId == null || lastPageId.length() == 0;
  82.                     }
  83.                 })
  84.                 .within(Time.seconds(5));
  86.         //3.把模式作用在流上
  87.         PatternStream<JSONObject> ps = CEP.pattern(keyedStream, pattern);
  88.         //4.取到匹配到的数据
  89.         SingleOutputStreamOperator<JSONObject> normal = ps.select(
  90.                 new OutputTag<JSONObject>("jump") {
  91.                 },
  92.                 new PatternTimeoutFunction<JSONObject, JSONObject>() {
  93.                     @Override
  94.                     public JSONObject timeout(Map<String, List<JSONObject>> map, long l) throws Exception {
  95.                         return map.get("entry1").get(0);
  96.                     }
  97.                 },
  98.                 //这个是正常访问的,不需要,可以不实现
  99.                 new PatternSelectFunction<JSONObject, JSONObject>() {
  100.                     @Override
  101.                     public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
  102.                         return map.get("entry1").get(0);
  103.                     }
  104.                 }
  105.         );
  106.         normal
  107.                 .union(normal.getSideOutput(new OutputTag<JSONObject>("jump") {}))
  108.                 .map(new MapFunction<JSONObject, String>() {
  109.                     @Override
  110.                     public String map(JSONObject value) throws Exception {
  111.                         return value.toJSONString();
  112.                     }
  113.                 })
  114.                 //.print();
  115.                 .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWM_UJ_DETAIL));
  118.     }
  119. }

3.测试

分别启动hadoop,zookeeper,kafka,gmall-logger,flink-yarn,dwdLogAPP,dwmUvApp,DwmUserJumpDetailApp

然后在终端起一个消费者消费 dwm_uj_detail

Flink实时数仓(二) - 图2

可以看到消费到的数据都是没有last_page_id的数据,即跳出数据

第四章.DWM层:订单宽表

订单是统计分析的重要对象,围绕订单有很多的维度统计需求,比如用户,地区,商品,品类,品牌等

为了之后统计计算更加方便,减少大表之间的关联,所以在实时计算中将围绕订单的相关数据整合成为一张订单的宽表

Flink实时数仓(二) - 图3

  1.     如上图,由于在之前的操作我们已经把数据分拆成了事实数据和维度数据,事实数据(订单表,订单明细)进入kafka数据流(DWD层中),维度数据进入Hbase中长期保存,那么我们在DWM层中需要把实时和维度数据进行整合关联在一起,形成宽表,那么这里就要处理有两种关联:事实数据与事实数据关联,事实数据与维度数据关联
  2.     1.事实数据与事实数据关联,其实就是流与流之间的关联
  3.     2.事实数据与维度数据关联,其实就是流计算中查询外部数据源

1.订单与订单明细关联

一.封装可以同时消费多个topic的BaseApp

  1. package com.atguigu.gmall.app;
  3. import com.atguigu.gmall.util.FlinkSourceUtil;
  4. import org.apache.flink.configuration.Configuration;
  5. import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
  6. import org.apache.flink.streaming.api.CheckpointingMode;
  7. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  8. import org.apache.flink.streaming.api.environment.CheckpointConfig;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. import java.util.ArrayList;
  12. import java.util.Arrays;
  13. import java.util.HashMap;
  15. public abstract class BaseAppV2 {
  16.     public abstract void run(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> stream);
  19.     public void init(int port,int p,String ck,String groupId,String topic,String... otherTopics){
  20.         System.setProperty("HADOOP_USER_NAME","atguigu");
  21.         Configuration conf = new Configuration();
  22.         conf.setInteger("rest.port",port);//设置web端端口号
  23.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
  24.         env.setParallelism(p);
  25.         //配置checkpoint
  26.         env.enableCheckpointing(3000, CheckpointingMode.EXACTLY_ONCE);
  27.         env.setStateBackend(new HashMapStateBackend());
  28.         env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop162:8020/gmall/" + ck);
  29.         env.getCheckpointConfig().setCheckpointTimeout(10 * 1000);
  30.         env.getCheckpointConfig()
  31.                 .enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  32.         ArrayList<String> topics = new ArrayList<>();
  33.         topics.add(topic);
  34.         topics.addAll(Arrays.asList(otherTopics));
  35.         HashMap<String, DataStreamSource<String>> topicAndStreamMap = new HashMap<>();
  36.         for (String t : topics) {
  37.             DataStreamSource<String> stream = env.addSource(FlinkSourceUtil.getKafkaSource(groupId, t));
  38.             topicAndStreamMap.put(t,stream);
  40.         }
  42.         //不同的APP有不同的业务逻辑
  43.         run(env,topicAndStreamMap);
  44.         try {
  45.             env.execute(ck);
  46.         } catch (Exception e) {
  47.             e.printStackTrace();
  48.         }
  49.     }
  50. }

二.实时表的join

为了取数据方便,封装几个pojo类

  1. OrderInfo.java
  1. package com.atguigu.gmall.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. import java.math.BigDecimal;
  8. import java.text.ParseException;
  9. import java.text.SimpleDateFormat;
  11. @Data
  12. @AllArgsConstructor
  13. @NoArgsConstructor
  14. public class OrderInfo {
  15.     private Long id;  // 订单id
  16.     private Long province_id;
  17.     private String order_status;
  18.     private Long user_id;
  19.     private BigDecimal total_amount; // 支付的金额
  20.     private BigDecimal activity_reduce_amount;
  21.     private BigDecimal coupon_reduce_amount;
  22.     private BigDecimal original_total_amount; // 没有优化和加运费之前的金额
  23.     private BigDecimal feight_fee;
  24.     private String expire_time;
  25.     private String create_time; // 年月日时分秒
  26.     private String operate_time;
  28.     private String create_date; // 把其他字段处理得到
  29.     private String create_hour;
  30.     private Long create_ts;
  32.     // 为了create_ts时间戳赋值, 所以需要手动补充
  33.     public void setCreate_time(String create_time) throws ParseException {
  34.         this.create_time = create_time;
  36.         this.create_date = this.create_time.substring(0, 10); // 年月日
  37.         this.create_hour = this.create_time.substring(11, 13); // 小时
  39.         final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  40.         this.create_ts = sdf.parse(create_time).getTime();
  42.     }
  44. }
  1. OrderDetail.java
  1. package com.atguigu.gmall.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. import java.math.BigDecimal;
  8. import java.text.ParseException;
  9. import java.text.SimpleDateFormat;
  11. @Data
  12. @AllArgsConstructor
  13. @NoArgsConstructor
  14. public class OrderDetail {
  15.     private Long id;
  16.     private Long order_id;
  17.     private Long sku_id;
  18.     private BigDecimal order_price;
  19.     private Long sku_num;
  20.     private String sku_name;
  21.     private String create_time;
  22.     private BigDecimal split_total_amount;
  23.     private BigDecimal split_activity_amount;
  24.     private BigDecimal split_coupon_amount;
  25.     private Long create_ts;
  26.     // 为了create_ts时间戳赋值, 所以需要手动补充
  27.     public void setCreate_time(String create_time) throws ParseException {
  28.         this.create_time = create_time;
  29.         final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  30.         this.create_ts = sdf.parse(create_time).getTime();
  32.     }
  34. }
  1. OrderWide.java
  1. package com.atguigu.gmall.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. import java.math.BigDecimal;
  8. import java.text.ParseException;
  9. import java.text.SimpleDateFormat;
  11. @Data
  12. @AllArgsConstructor
  13. @NoArgsConstructor
  14. public class OrderWide {
  15.     private Long detail_id;
  16.     private Long order_id;
  17.     private Long sku_id;
  18.     private BigDecimal order_price;
  19.     private Long sku_num;
  20.     private String sku_name;
  21.     private Long province_id;
  22.     private String order_status;
  23.     private Long user_id;
  25.     private BigDecimal total_amount;
  26.     private BigDecimal activity_reduce_amount;
  27.     private BigDecimal coupon_reduce_amount;
  28.     private BigDecimal original_total_amount;
  29.     private BigDecimal feight_fee;
  30.     private BigDecimal split_feight_fee;
  31.     private BigDecimal split_activity_amount;
  32.     private BigDecimal split_coupon_amount;
  33.     private BigDecimal split_total_amount;
  35.     private String expire_time;
  36.     private String create_time;
  37.     private String operate_time;
  38.     private String create_date; // 把其他字段处理得到
  39.     private String create_hour;
  41.     private String province_name;//查询维表得到
  42.     private String province_area_code;
  43.     private String province_iso_code;
  44.     private String province_3166_2_code;
  46.     private Integer user_age;
  47.     private String user_gender;
  49.     private Long spu_id;     //作为维度数据 要关联进来
  50.     private Long tm_id;
  51.     private Long category3_id;
  52.     private String spu_name;
  53.     private String tm_name;
  54.     private String category3_name;
  56.     public OrderWide(OrderInfo orderInfo, OrderDetail orderDetail) {
  57.         mergeOrderInfo(orderInfo);
  58.         mergeOrderDetail(orderDetail);
  60.     }
  62.     public void mergeOrderInfo(OrderInfo orderInfo) {
  63.         if (orderInfo != null) {
  64.             this.order_id = orderInfo.getId();
  65.             this.order_status = orderInfo.getOrder_status();
  66.             this.create_time = orderInfo.getCreate_time();
  67.             this.create_date = orderInfo.getCreate_date();
  68.             this.create_hour = orderInfo.getCreate_hour();
  69.             this.activity_reduce_amount = orderInfo.getActivity_reduce_amount();
  70.             this.coupon_reduce_amount = orderInfo.getCoupon_reduce_amount();
  71.             this.original_total_amount = orderInfo.getOriginal_total_amount();
  72.             this.feight_fee = orderInfo.getFeight_fee();
  73.             this.total_amount = orderInfo.getTotal_amount();
  74.             this.province_id = orderInfo.getProvince_id();
  75.             this.user_id = orderInfo.getUser_id();
  76.         }
  77.     }
  79.     public void mergeOrderDetail(OrderDetail orderDetail) {
  80.         if (orderDetail != null) {
  81.             this.detail_id = orderDetail.getId();
  82.             this.sku_id = orderDetail.getSku_id();
  83.             this.sku_name = orderDetail.getSku_name();
  84.             this.order_price = orderDetail.getOrder_price();
  85.             this.sku_num = orderDetail.getSku_num();
  86.             this.split_activity_amount = orderDetail.getSplit_activity_amount();
  87.             this.split_coupon_amount = orderDetail.getSplit_coupon_amount();
  88.             this.split_total_amount = orderDetail.getSplit_total_amount();
  89.         }
  90.     }
  92.     public void calcUserAge(String birthday) throws ParseException {
  93.         long now = System.currentTimeMillis();
  94.         long birth = new SimpleDateFormat("yyyy-MM-dd").parse(birthday).getTime();
  96.         this.user_age = (int) ((now - birth) / 1000 / 60 / 60 / 24 / 365);
  97.     }
  98. }
  1. 实时表的join
  1.  /**
  2.      * 事实表的join(orderDetail和orderInfo)
  3.      * @param streams
  4.      */
  5.     private SingleOutputStreamOperator<OrderWide> joinFacts(HashMap<String, DataStreamSource<String>> streams) {
  6.         KeyedStream<OrderInfo, Long> orderInfoStream = streams.get(Constant.TOPIC_DWD_ORDER_INFO)
  7.                 .map(new MapFunction<String, OrderInfo>() {
  8.                     @Override
  9.                     public OrderInfo map(String value) throws Exception {
  10.                         return JSON.parseObject(value, OrderInfo.class);
  11.                     }
  12.                 })
  13.                 .assignTimestampsAndWatermarks(
  14.                         WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  15.                                 .withTimestampAssigner((info, ts) -> info.getCreate_ts())
  16.                 )
  17.                 .keyBy(orderInfo -> orderInfo.getId());
  18.         KeyedStream<OrderDetail, Long> orderDetailLongKeyedStream = streams.get(Constant.TOPIC_DWD_ORDER_DETAIL)
  19.                 .map(new MapFunction<String, OrderDetail>() {
  20.                     @Override
  21.                     public OrderDetail map(String value) throws Exception {
  22.                         return JSON.parseObject(value, OrderDetail.class);
  23.                     }
  24.                 })
  25.                 .assignTimestampsAndWatermarks(
  26.                         WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  27.                                 .withTimestampAssigner((info, ts) -> info.getCreate_ts())
  28.                 )
  29.                 .keyBy(orderDetail -> orderDetail.getOrder_id());
  30.         return orderInfoStream
  31.                 .intervalJoin(orderDetailLongKeyedStream)
  32.                 .between(Time.seconds(-10),Time.seconds(10))
  33.                 .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
  34.                     @Override
  35.                     public void processElement(OrderInfo left, OrderDetail right, Context ctx, Collector<OrderWide> out) throws Exception {
  36.                         out.collect(new OrderWide(left,right));
  37.                     }
  38.                 });
  40.     }

2.关联维度表

维度关联实际上就是在流中查询存储在hbase中的数据表

但是即使通过主键的方式查询,hbase速度的查询也是不及流之间的join,外部数据源的查询常常是流式计算的性能瓶颈,所以在这个基础上还能进行一定的优化

一.将维度数据初始化到Hbase

需要开启hbase,Maxwell,DwdDbApp

共用到六张维度表: user_info,base_province,sku_info,spu_info,base_category3,base_trademark

  1. bin/maxwell-bootstrap --user root  --password aaaaaa --host hadoop162  --database gmall2021 --table user_info --client_id maxwell_1
  3. bin/maxwell-bootstrap --user root  --password aaaaaa --host hadoop162  --database gmall2021 --table base_province --client_id maxwell_1
  5. bin/maxwell-bootstrap --user root  --password aaaaaa --host hadoop162  --database gmall2021 --table sku_info --client_id maxwell_1
  7. bin/maxwell-bootstrap --user root  --password aaaaaa --host hadoop162  --database gmall2021 --table spu_info --client_id maxwell_1
  9. bin/maxwell-bootstrap --user root  --password aaaaaa --host hadoop162  --database gmall2021 --table base_category3 --client_id maxwell_1
  11. bin/maxwell-bootstrap --user root  --password aaaaaa --host hadoop162  --database gmall2021 --table base_trademark --client_id maxwell_1

二.更新JdbcUtil工具类

实现从Jdbc中查询数据

  1. package com.atguigu.gmall.util;
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.atguigu.gmall.common.Constant;
  5. import org.apache.commons.beanutils.BeanUtils;
  7. import java.sql.*;
  8. import java.util.ArrayList;
  9. import java.util.List;
  11. public class JdbcUtil {
  12.     public static Connection getPhoenixConnection(String phoenixUrl) throws SQLException, ClassNotFoundException {
  13.         String phoenixDriver = Constant.PHOENIX_DRIVER;
  14.         return getJdbcConnection(phoenixDriver,phoenixUrl);
  15.     }
  17.     private static Connection getJdbcConnection(String driver, String url) throws ClassNotFoundException, SQLException {
  18.         Class.forName(driver);
  19.         return DriverManager.getConnection(url);
  20.     }
  22.     public static void main(String[] args) throws SQLException, ClassNotFoundException {
  23.         List<JSONObject> list = queryList(getPhoenixConnection(Constant.PHOENIX_URL),
  24.                 "select * from dim_user_info where id=?",
  25.                 new String[]{"1"},
  26.                 JSONObject.class
  28.         );
  29.         for (JSONObject jsonObject : list) {
  30.             System.out.println(jsonObject);
  32.         }
  33.     }
  36.     public static <T> List<T> queryList(Connection conn,
  37.                                         String sql,
  38.                                         Object[] args,
  39.                                         Class<T> tClass ){
  40.         ArrayList<T> result = new ArrayList<>();
  41.         //查询数据库
  42.         try {
  43.             PreparedStatement ps = conn.prepareStatement(sql);
  44.             //给SQL中的占位符进行赋值
  45.             for (int i = 0; args != null && i < args.length; i++) {
  46.                 ps.setObject(i+1,args[i]);
  48.             }
  49.             ResultSet resultSet = ps.executeQuery();
  50.             ResultSetMetaData metaData = resultSet.getMetaData();
  51.             while(resultSet.next()){
  52.                 //创建一个T类型的对象,把每列的数据都封装在T对象中
  53.                 T t = tClass.newInstance();
  54.                 for (int i = 1; i < metaData.getColumnCount(); i++) {
  55.                     //遍历每列数据
  56.                     String columnName = metaData.getColumnName(i);
  57.                     Object value = resultSet.getObject(columnName);
  58.                     BeanUtils.setProperty(t,columnName,value);
  60.                 }
  61.                 result.add(t);
  62.             }
  63.         } catch (Exception e) {
  64.             e.printStackTrace();
  65.         }
  66.         return result;
  68.     }
  69. }

三.封装DimUtil类,方便从Phoenix中获取数据

  1. package com.atguigu.gmall.util;
  3. import com.alibaba.fastjson.JSONObject;
  5. import java.sql.Connection;
  6. import java.util.List;
  8. public class DimUtil {
  9.     public static JSONObject getDimFromPhoenix(Connection phoenixConn,
  10.                                                String table,
  11.                                                Long id){
  12.         String sql = "select * from " + table + " where id=?";
  13.         Object[] args = {id.toString()};
  14.         List<JSONObject> result = JdbcUtil.queryList(phoenixConn, sql, args, JSONObject.class);
  15.         return result.size()==1 ? result.get(0): null;
  16.     }
  17. }

四.Join维度表代码

  1. package com.atguigu.gmall.app.dwm;
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.JSONObject;
  5. import com.atguigu.gmall.app.BaseAppV2;
  6. import com.atguigu.gmall.common.Constant;
  7. import com.atguigu.gmall.pojo.OrderDetail;
  8. import com.atguigu.gmall.pojo.OrderInfo;
  9. import com.atguigu.gmall.pojo.OrderWide;
  10. import com.atguigu.gmall.util.DimUtil;
  11. import com.atguigu.gmall.util.JdbcUtil;
  12. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  13. import org.apache.flink.api.common.functions.MapFunction;
  14. import org.apache.flink.api.common.functions.RichMapFunction;
  15. import org.apache.flink.configuration.Configuration;
  16. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  17. import org.apache.flink.streaming.api.datastream.KeyedStream;
  18. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  19. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  21. import java.sql.Connection;
  22. import java.time.Duration;
  23. import java.util.HashMap;
  24. import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
  25. import org.apache.flink.streaming.api.windowing.time.Time;
  26. import org.apache.flink.util.Collector;
  28. public class DwmOrderWriteApp extends BaseAppV2 {
  29.     public static void main(String[] args) {
  30.         new DwmOrderWriteApp().init(3003,1,"DwmOrderWriteApp","DwmOrderWriteApp", Constant.TOPIC_DWD_ORDER_INFO,Constant.TOPIC_DWD_ORDER_DETAIL);
  31.     }
  32.     @Override
  33.     public void run(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streams) {
  34.         //streams.get(Constant.TOPIC_DWD_ORDER_INFO).print(Constant.TOPIC_DWD_ORDER_INFO);
  35.         //streams.get(Constant.TOPIC_DWD_ORDER_DETAIL).print(Constant.TOPIC_DWD_ORDER_DETAIL);
  36.         //1.事实表的join: interval join
  37.         SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims = joinFacts(streams);
  40.         //2.join维度数据
  41.         joinDims(orderWideStreamWithoutDims);
  45.         //3.把结果写入到kafka中,给dws准备数据
  48.     }
  50.     private void joinDims(SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims) {
  51.         //join 6 张维度表
  52.         //根据维度表中的某个id去查找对应的那一条数据
  53.         orderWideStreamWithoutDims
  54.                 .map(new RichMapFunction<OrderWide, OrderWide>() {
  56.                     private Connection phoenixConnection;
  58.                     @Override
  59.                     public void open(Configuration parameters) throws Exception {
  60.                         phoenixConnection = JdbcUtil.getPhoenixConnection(Constant.PHOENIX_URL);
  61.                     }
  63.                     @Override
  64.                     public OrderWide map(OrderWide orderWide) throws Exception {
  65.                        //1.补充user
  66.                         JSONObject userInfo = DimUtil.getDimFromPhoenix(
  67.                                 phoenixConnection,
  68.                                 "dim_user_info",
  69.                                 orderWide.getUser_id());
  70.                         orderWide.setUser_gender(userInfo.getString("GENDER"));
  71.                         orderWide.calcUserAge(userInfo.getString("BIRTHDAY"));
  73.                         // 2. 补齐省份
  74.                         JSONObject provinceInfo = DimUtil.getDimFromPhoenix(phoenixConnection,
  75.                                 "dim_base_province",
  76.                                 orderWide.getProvince_id());
  77.                         orderWide.setProvince_name(provinceInfo.getString("NAME"));
  78.                         orderWide.setProvince_iso_code(provinceInfo.getString("ISO_CODE"));
  79.                         orderWide.setProvince_area_code(provinceInfo.getString("AREA_CODE"));
  80.                         orderWide.setProvince_3166_2_code(provinceInfo.getString("ISO_3166_2"));
  82.                         // 3. sku
  83.                         JSONObject skuInfo = DimUtil.getDimFromPhoenix(phoenixConnection,
  84.                                 "dim_sku_info",
  85.                                 orderWide.getSku_id());
  86.                         orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
  88.                         orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
  89.                         orderWide.setTm_id(skuInfo.getLong("TM_ID"));
  90.                         orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
  92.                         // 4. spu
  93.                         JSONObject spuInfo = DimUtil.getDimFromPhoenix(phoenixConnection,
  94.                                 "dim_spu_info",
  95.                                 orderWide.getSpu_id());
  96.                         orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));
  99.                         // 5. tm
  100.                         JSONObject tmInfo = DimUtil.getDimFromPhoenix(phoenixConnection,
  101.                                 "dim_base_trademark",
  102.                                 orderWide.getTm_id());
  103.                         orderWide.setTm_name(tmInfo.getString("TM_NAME"));
  105.                         // 5. tm
  106.                         JSONObject c3Info = DimUtil.getDimFromPhoenix(phoenixConnection,
  107.                                 "dim_base_category3",
  108.                                 orderWide.getCategory3_id());
  109.                         orderWide.setCategory3_name(c3Info.getString("NAME"));
  111.                         return orderWide;
  112.                     }
  113.                 })
  114.                 .print();
  115.     }
  117.     /**
  118.      * 事实表的join(orderDetail和orderInfo)
  119.      * @param streams
  120.      */
  121.     private SingleOutputStreamOperator<OrderWide> joinFacts(HashMap<String, DataStreamSource<String>> streams) {
  122.         KeyedStream<OrderInfo, Long> orderInfoStream = streams.get(Constant.TOPIC_DWD_ORDER_INFO)
  123.                 .map(new MapFunction<String, OrderInfo>() {
  124.                     @Override
  125.                     public OrderInfo map(String value) throws Exception {
  126.                         return JSON.parseObject(value, OrderInfo.class);
  127.                     }
  128.                 })
  129.                 .assignTimestampsAndWatermarks(
  130.                         WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  131.                                 .withTimestampAssigner((info, ts) -> info.getCreate_ts())
  132.                 )
  133.                 .keyBy(orderInfo -> orderInfo.getId());
  134.         KeyedStream<OrderDetail, Long> orderDetailLongKeyedStream = streams.get(Constant.TOPIC_DWD_ORDER_DETAIL)
  135.                 .map(new MapFunction<String, OrderDetail>() {
  136.                     @Override
  137.                     public OrderDetail map(String value) throws Exception {
  138.                         return JSON.parseObject(value, OrderDetail.class);
  139.                     }
  140.                 })
  141.                 .assignTimestampsAndWatermarks(
  142.                         WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  143.                                 .withTimestampAssigner((info, ts) -> info.getCreate_ts())
  144.                 )
  145.                 .keyBy(orderDetail -> orderDetail.getOrder_id());
  146.         return orderInfoStream
  147.                 .intervalJoin(orderDetailLongKeyedStream)
  148.                 .between(Time.seconds(-10),Time.seconds(10))
  149.                 .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
  150.                     @Override
  151.                     public void processElement(OrderInfo left, OrderDetail right, Context ctx, Collector<OrderWide> out) throws Exception {
  152.                         out.collect(new OrderWide(left,right));
  153.                     }
  154.                 });
  156.     }
  157. }

五.启动APP进行测试,发现维度信息已经添加成功

Flink实时数仓(二) - 图4

3.维度管理优化一:加入缓存

上一节实现的功能是直接查询的Hbase,外部数据源的查询往往是流式计算的性能瓶颈,所以我们需要在之前实现的基础上进行一定的优化,使用redis缓存

一.为什么使用redis缓存

  1.     要实现的目标:第一次从数据库读,这条以后应该从缓存(内存)
  2.     1.把维度数据缓存到flink的状态中
  3.         优点:本地内存,快,不需要网络,数据结构也比较丰富
  4.         缺点: 
  5.             一.对flink的内存有压力
  6.             二.维度有变化,缓存的数据没有办法收到这个变化,因为缓存是在dwm层的应用,维度数据是dwdDbApp负责写入到hbase的,如果有变化只有这个APP知道
  7.     2.把维度数据缓存到外部专用的缓存:redis
  8.         优点:专用缓存,容器比较大,速度也快,如果维度发生变化,dwdDbApp可以直接访问Redis去更新缓存
  9.         缺点:需要用过网络
  10.     3.综合考虑,就应该使用redis缓存

二.redis中数据结构的选择

数据结构 key value 缺点 优点
String 表名加Id(dim_user_info:1) json格式字串 key比较多,管理不方便 1.读写方便2.可以单独给每个key设置过期时间
list 表名(dim_user_info) 每个数据的json格式 读很难 一张表一个key,写方便
set 表名(dim_user_info) 每个数据的json格式 读很难 一张表一个key,写方便
hash 表名 field+value 没有办法单独给每条数据设置过期时间 一个表一个字段,读写也方便

综合考虑使用String结构比较合适

三.封装Redis客户端工具类

  1. 导入Jedis依赖
  1. <dependency>
  2.     <groupId>redis.clients</groupId>
  3.     <artifactId>jedis</artifactId>
  4.     <version>3.2.0</version>
  5. </dependency>
  1. 封装Redis工具类
  1. package com.atguigu.gmall.util;
  3. import redis.clients.jedis.Jedis;
  4. import redis.clients.jedis.JedisPool;
  5. import redis.clients.jedis.JedisPoolConfig;
  7. public class RedisUtil {
  8.     private static JedisPool pool;
  9.     static {
  10.         JedisPoolConfig conf = new JedisPoolConfig();
  11.         conf.setMaxTotal(300);
  12.         conf.setMaxIdle(10);
  13.         conf.setMaxWaitMillis(10 * 1000);
  14.         conf.setMinIdle(5);
  15.         conf.setTestOnBorrow(true);
  16.         conf.setTestOnCreate(true);
  17.         conf.setTestOnReturn(true);
  18.         pool = new JedisPool(conf,"hadoop162",6379);
  19.     }
  20.     public static Jedis getRedisClient(){
  21.         return pool.getResource();
  22.     }
  23. }

四.使用缓存读取维度数据

  1. DimUtil.java
  1. package com.atguigu.gmall.util;
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.JSONObject;
  5. import redis.clients.jedis.Jedis;
  7. import java.sql.Connection;
  8. import java.util.List;
  10. public class DimUtil {
  11.     public static JSONObject getDimFromPhoenix(Connection phoenixConn,
  12.                                                String table,
  13.                                                Long id){
  14.         String sql = "select * from " + table + " where id=?";
  15.         Object[] args = {id.toString()};
  16.         List<JSONObject> result = JdbcUtil.queryList(phoenixConn, sql, args, JSONObject.class);
  17.         return result.size()==1 ? result.get(0): null;
  18.     }
  20.     public static JSONObject getDim(Connection phoenixConnection,
  21.                                     Jedis redisClient,
  22.                                     String table,
  23.                                     Long id) {
  24.         //1.先从redis中获取
  25.         JSONObject dim = getDimFromRedis(redisClient, table, id);
  26.         //3.获取不到从Phoenix中获取,结果返回,把结果写入到redis
  27.         if(dim == null){
  28.             System.out.println(table + " " + id + "走的数据库");
  29.             dim = getDimFromPhoenix(phoenixConnection,table,id);
  30.             //把维度数据写入到redis中
  31.             writeToRedis(redisClient,table,id,dim);
  33.         }else{
  34.             System.out.println(table + " " + id + "走的缓存");
  35.         }
  37.         //2.获取到了信息,直接返回
  38.         return dim;
  39.     }
  41.     private static void writeToRedis(Jedis redisClient,
  42.                                      String table,
  43.                                      Long id,
  44.                                      JSONObject dim) {
  45.         String key = table + ":" + id;
  46.         String value = dim.toJSONString();
  47.         redisClient.setex(key,3*24*60*60,value);
  48.     }
  50.     private static JSONObject getDimFromRedis(Jedis redisClient, String table, Long id) {
  51.         String key = table + ":" + id;
  52.         String dim = redisClient.get(key); //如果缓存不存在,就是null
  53.         return dim == null ? null : JSON.parseObject(dim);
  54.     }
  55. }
  1. DwmOrderWriteApp_Cache .java
  1. package com.atguigu.gmall.app.dwm;
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.JSONObject;
  5. import com.atguigu.gmall.app.BaseAppV2;
  6. import com.atguigu.gmall.common.Constant;
  7. import com.atguigu.gmall.pojo.OrderDetail;
  8. import com.atguigu.gmall.pojo.OrderInfo;
  9. import com.atguigu.gmall.pojo.OrderWide;
  10. import com.atguigu.gmall.util.DimUtil;
  11. import com.atguigu.gmall.util.JdbcUtil;
  12. import com.atguigu.gmall.util.RedisUtil;
  13. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  14. import org.apache.flink.api.common.functions.MapFunction;
  15. import org.apache.flink.api.common.functions.RichMapFunction;
  16. import org.apache.flink.configuration.Configuration;
  17. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  18. import org.apache.flink.streaming.api.datastream.KeyedStream;
  19. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  20. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  21. import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
  22. import org.apache.flink.streaming.api.windowing.time.Time;
  23. import org.apache.flink.util.Collector;
  24. import redis.clients.jedis.Jedis;
  26. import java.sql.Connection;
  27. import java.time.Duration;
  28. import java.util.HashMap;
  30. public class DwmOrderWriteApp_Cache extends BaseAppV2 {
  31.     public static void main(String[] args) {
  32.         new DwmOrderWriteApp_Cache().init(3004,1,"DwmOrderWriteApp_Cache","DwmOrderWriteApp_Cache", Constant.TOPIC_DWD_ORDER_INFO,Constant.TOPIC_DWD_ORDER_DETAIL);
  33.     }
  34.     @Override
  35.     public void run(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streams) {
  36.         //streams.get(Constant.TOPIC_DWD_ORDER_INFO).print(Constant.TOPIC_DWD_ORDER_INFO);
  37.         //streams.get(Constant.TOPIC_DWD_ORDER_DETAIL).print(Constant.TOPIC_DWD_ORDER_DETAIL);
  38.         //1.事实表的join: interval join
  39.         SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims = joinFacts(streams);
  42.         //2.join维度数据
  43.         joinDims(orderWideStreamWithoutDims);
  47.         //3.把结果写入到kafka中,给dws准备数据
  50.     }
  52.     private void joinDims(SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims) {
  53.         //join 6 张维度表
  54.         //根据维度表中的某个id去查找对应的那一条数据
  55.         orderWideStreamWithoutDims
  56.                 .map(new RichMapFunction<OrderWide, OrderWide>() {
  57.                     @Override
  58.                     public void close() throws Exception {
  59.                         if(phoenixConnection != null){
  60.                             phoenixConnection.close();
  61.                         }
  62.                         if(redisClient != null){
  63.                             //如果客户端是new Jedis()这样出来,则是关闭客户端,
  64.                             // 如果是通过连接池获取的,则归还给连接池
  65.                             redisClient.close();
  66.                         }
  67.                     }
  69.                     private Jedis redisClient;
  70.                     private Connection phoenixConnection;
  72.                     @Override
  73.                     public void open(Configuration parameters) throws Exception {
  74.                         phoenixConnection = JdbcUtil.getPhoenixConnection(Constant.PHOENIX_URL);
  75.                         //获取redis客户端
  76.                         redisClient = RedisUtil.getRedisClient();
  77.                     }
  79.                     @Override
  80.                     public OrderWide map(OrderWide orderWide) throws Exception {
  81.                        //1.补充user
  82.                         JSONObject userInfo = DimUtil.getDim(
  83.                                 phoenixConnection,
  84.                                 redisClient,
  85.                                 "dim_user_info",
  86.                                 orderWide.getUser_id());
  87.                         orderWide.setUser_gender(userInfo.getString("GENDER"));
  88.                         orderWide.calcUserAge(userInfo.getString("BIRTHDAY"));
  90.                         // 2. 补齐省份
  91.                         JSONObject provinceInfo = DimUtil.getDim(
  92.                                 phoenixConnection,
  93.                                 redisClient,
  94.                                 "dim_base_province",
  95.                                 orderWide.getProvince_id());
  96.                         orderWide.setProvince_name(provinceInfo.getString("NAME"));
  97.                         orderWide.setProvince_iso_code(provinceInfo.getString("ISO_CODE"));
  98.                         orderWide.setProvince_area_code(provinceInfo.getString("AREA_CODE"));
  99.                         orderWide.setProvince_3166_2_code(provinceInfo.getString("ISO_3166_2"));
  101.                         // 3. sku
  102.                         JSONObject skuInfo = DimUtil.getDim(
  103.                                 phoenixConnection,
  104.                                 redisClient,
  105.                                 "dim_sku_info",
  106.                                 orderWide.getSku_id());
  107.                         orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
  109.                         orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
  110.                         orderWide.setTm_id(skuInfo.getLong("TM_ID"));
  111.                         orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
  113.                         // 4. spu
  114.                         JSONObject spuInfo = DimUtil.getDim(
  115.                                 phoenixConnection,
  116.                                 redisClient,
  117.                                 "dim_spu_info",
  118.                                 orderWide.getSpu_id());
  119.                         orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));
  122.                         // 5. tm
  123.                         JSONObject tmInfo =DimUtil.getDim(
  124.                                 phoenixConnection,
  125.                                 redisClient,
  126.                                 "dim_base_trademark",
  127.                                 orderWide.getTm_id());
  128.                         orderWide.setTm_name(tmInfo.getString("TM_NAME"));
  130.                         // 5. tm
  131.                         JSONObject c3Info =DimUtil.getDim(
  132.                                 phoenixConnection,
  133.                                 redisClient,
  134.                                 "dim_base_category3",
  135.                                 orderWide.getCategory3_id());
  136.                         orderWide.setCategory3_name(c3Info.getString("NAME"));
  138.                         return orderWide;
  139.                     }
  140.                 })
  141.                 .print();
  142.     }
  144.     /**
  145.      * 事实表的join(orderDetail和orderInfo)
  146.      * @param streams
  147.      */
  148.     private SingleOutputStreamOperator<OrderWide> joinFacts(HashMap<String, DataStreamSource<String>> streams) {
  149.         KeyedStream<OrderInfo, Long> orderInfoStream = streams.get(Constant.TOPIC_DWD_ORDER_INFO)
  150.                 .map(new MapFunction<String, OrderInfo>() {
  151.                     @Override
  152.                     public OrderInfo map(String value) throws Exception {
  153.                         return JSON.parseObject(value, OrderInfo.class);
  154.                     }
  155.                 })
  156.                 .assignTimestampsAndWatermarks(
  157.                         WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  158.                                 .withTimestampAssigner((info, ts) -> info.getCreate_ts())
  159.                 )
  160.                 .keyBy(orderInfo -> orderInfo.getId());
  161.         KeyedStream<OrderDetail, Long> orderDetailLongKeyedStream = streams.get(Constant.TOPIC_DWD_ORDER_DETAIL)
  162.                 .map(new MapFunction<String, OrderDetail>() {
  163.                     @Override
  164.                     public OrderDetail map(String value) throws Exception {
  165.                         return JSON.parseObject(value, OrderDetail.class);
  166.                     }
  167.                 })
  168.                 .assignTimestampsAndWatermarks(
  169.                         WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  170.                                 .withTimestampAssigner((info, ts) -> info.getCreate_ts())
  171.                 )
  172.                 .keyBy(orderDetail -> orderDetail.getOrder_id());
  173.         return orderInfoStream
  174.                 .intervalJoin(orderDetailLongKeyedStream)
  175.                 .between(Time.seconds(-10),Time.seconds(10))
  176.                 .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
  177.                     @Override
  178.                     public void processElement(OrderInfo left, OrderDetail right, Context ctx, Collector<OrderWide> out) throws Exception {
  179.                         out.collect(new OrderWide(left,right));
  180.                     }
  181.                 });
  183.     }
  184. }
  1. 启动APP测试

Flink实时数仓(二) - 图5

Flink实时数仓(二) - 图6

可以看到,已经可以从redis中获取数据了

五.在DwdDbApp中更新缓存

  1. PhoenixSink.java
  1. package com.atguigu.gmall.sink;
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.atguigu.gmall.common.Constant;
  5. import com.atguigu.gmall.pojo.TableProcess;
  6. import com.atguigu.gmall.util.JdbcUtil;
  7. import com.atguigu.gmall.util.RedisUtil;
  8. import org.apache.flink.api.common.state.ValueState;
  9. import org.apache.flink.api.common.state.ValueStateDescriptor;
  10. import org.apache.flink.api.java.tuple.Tuple2;
  11. import org.apache.flink.configuration.Configuration;
  12. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  13. import org.apache.flink.streaming.api.functions.sink.SinkFunction;
  14. import redis.clients.jedis.Jedis;
  16. import java.io.IOException;
  17. import java.sql.Connection;
  18. import java.sql.PreparedStatement;
  19. import java.sql.SQLException;
  20. import java.util.Map;
  22. public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
  23.     private Connection conn;
  24.     private ValueState<Boolean> isFirst;
  25.     private Jedis redisClient;
  27.     /**
  28.      * 建立到Phoenix的连接
  29.      * @param parameters
  30.      * @throws Exception
  31.      */
  32.     @Override
  33.     public void open(Configuration parameters) throws Exception {
  34.         String url = Constant.PHOENIX_URL;
  35.         conn = JdbcUtil.getPhoenixConnection(url);
  36.         isFirst = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isFirst",Boolean.class));
  37.         redisClient = RedisUtil.getRedisClient();
  38.     }
  40.     /**
  41.      * 关闭连接
  42.      * @throws Exception
  43.      */
  44.     @Override
  45.     public void close() throws Exception {
  46.         if(conn != null){
  47.             conn.close();
  48.         }
  49.     }
  53.     @Override
  54.     public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {
  55.         //1.建表
  56.         checkTable(value);
  57.         //2.写数据
  58.         writeData(value);
  59.         //3.更新缓存
  60.         updateCache(value);
  61.     }
  63.     private void updateCache(Tuple2<JSONObject, TableProcess> value) {
  64.         JSONObject data = value.f0;
  65.         TableProcess tp = value.f1;
  66.         String key = tp.getSink_table() + ":" + data.getLong("id");
  67.         //这次操作是更新维度还是去redis中存这个缓存
  68.         if("update".equals(tp.getOperate_type()) && redisClient.exists(key)){
  69.             //应该把字段名全部大写
  70.             JSONObject upperDim = new JSONObject();
  71.             for (Map.Entry<String, Object> entry : data.entrySet()) {
  72.                 upperDim.put(entry.getKey().toUpperCase(),entry.getValue());
  74.             }
  75.             redisClient.set(key, upperDim.toString());
  76.         }
  77.     }
  79.     private void writeData(Tuple2<JSONObject, TableProcess> value) throws SQLException {
  80.         JSONObject data = value.f0;
  81.         TableProcess tp = value.f1;
  82.         //拼接插入语句
  83.         //upsert into t(a,b,c) values(?,?,?)
  84.         StringBuilder sql = new StringBuilder();
  85.         sql.append("upsert into ")
  86.                 .append(tp.getSink_table())
  87.                 .append("(")
  88.                 .append(tp.getSink_columns())
  89.                 .append(")values(")
  90.                 .append(tp.getSink_columns().replaceAll("[^,]+", "?"))
  91.                 .append(")");
  92.         PreparedStatement ps = conn.prepareStatement(sql.toString());
  93.         //给占位符赋值,根据列名去data中取值
  94.         String[] cs = tp.getSink_columns().split(",");
  95.         for (int i = 0; i < cs.length; i++) {
  96.             String columnName = cs[i];
  97.             Object v = data.get(columnName);
  98.             ps.setString(i+1,v==null?null:v.toString());
  99.         }
  102.         //执行SQL
  103.         ps.execute();
  104.         conn.commit();
  105.         ps.close();
  106.     }
  108.     /**
  109.      * 在Phoenix中进行建表
  110.      * 执行建表语句
  111.      * @param value
  112.      */
  113.     private void checkTable(Tuple2<JSONObject, TableProcess> value) throws IOException, SQLException {
  114.         if(isFirst.value() == null){
  115.             TableProcess tp = value.f1;
  116.             StringBuilder sql = new StringBuilder();
  117.             sql
  118.                     .append("create table if not exists ")
  119.                     .append(tp.getSink_table())
  120.                     .append("(")
  121.                     .append(tp.getSink_columns().replaceAll(",", " varchar, "))
  122.                     .append(" varchar, constraint pk primary key(")
  123.                     .append(tp.getSink_pk() == null? "id" : tp.getSink_pk())
  124.                     .append("))")
  125.                     .append(tp.getSink_extend() == null ? "" : tp.getSink_extend());
  126.             System.out.println(sql.toString());
  127.             PreparedStatement ps = conn.prepareStatement(sql.toString());
  128.             ps.execute();
  129.             conn.commit();
  130.             ps.close();
  131.             isFirst.update(true);
  133.         }
  134.     }
  135. }
  1. 测试(重新将程序打包上传到linux上并启动)

更改业务数据之前redis里的userinfo

Flink实时数仓(二) - 图7

将数据库中对应的姓名修改为黄蓉

Flink实时数仓(二) - 图8

再从redis中查看,发现姓名已经修改成功

Flink实时数仓(二) - 图9

4.维度管理优化二:使用异步查询

一.异步IO的介绍

  1.     在Flink 流处理过程中，经常需要和外部系统进行交互，用维度表补全事实表中的字段。
  2.     例如：在电商场景中，需要一个商品的skuid去关联商品的一些属性，例如商品所属行业、商品的生产厂家、生产厂家的一些情况；在物流场景中，知道包裹id，需要去关联包裹的行业属性、发货信息、收货信息等等。
  3.     默认情况下，在Flink的MapFunction中，单个并行只能用同步方式去交互: 将请求发送到外部存储，IO阻塞，等待请求返回，然后继续发送下一个请求。这种同步交互的方式往往在网络等待上就耗费了大量时间。为了提高处理效率，可以增加MapFunction的并行度，但增加并行度就意味着更多的资源，并不是一种非常好的解决方式。
  4.     Flink 在1.2中引入了Async I/O，在异步模式下，将IO操作异步化，单个并行可以连续发送多个请求，哪个请求先返回就先处理，从而在连续的请求间不需要阻塞式等待，大大提高了流处理效率。
  5.     Async I/O 是阿里巴巴贡献给社区的一个呼声非常高的特性，解决与外部系统交互时网络延迟成为了系统瓶颈的问题。
  6.     异步查询实际上是把维表的查询操作托管给单独的线程池完成，这样不会因为某一个查询造成阻塞，单个并行可以连续发送多个请求，提高并发效率。
  7.     这种方式特别针对涉及网络IO的操作，减少因为请求等待带来的消耗。

Flink实时数仓(二) - 图10

使用异步API的先决条件

  • 正确地实现数据库（或键/值存储）的异步 I/O 交互需要支持异步请求的数据库客户端。许多主流数据库都提供了这样的客户端。
  • 如果没有这样的客户端，可以通过创建多个客户端并使用线程池处理同步调用的方法，将同步客户端转换为有限并发的客户端。然而，这种方法通常比正规的异步客户端效率低。
  • Phoenix目前没有提供异步的客户端, 所以只能通过创建多个客户端并使用线程池处理同步调用的方法，将同步客户端转换为有限并发的客户端

Flink的异步I/O API

Flink 的异步 I/O API 允许用户在流处理中使用异步请求客户端。API 处理与数据流的集成，同时还能处理好顺序、事件时间和容错等。

在具备异步数据库客户端的基础上，实现数据流转换操作与数据库的异步 I/O 交互需要以下三部分：

  1. 实现分发请求的 AsyncFunction
  2. 获取数据库交互的结果并发送给ResultFuture的回调函数
  3. 将异步API操作应用于DataStream作为DataStream的一次转换操作

二.封装线程池工具类

  1. package com.atguigu.gmall.util;
  3. import java.util.concurrent.LinkedBlockingDeque;
  4. import java.util.concurrent.ThreadPoolExecutor;
  5. import java.util.concurrent.TimeUnit;
  7. /**
  8.  * 用于异步的线程池工具类
  9.  */
  11. public class ThreadPoolUtil {
  12.     public static ThreadPoolExecutor getThreadPool(){
  13.         return new ThreadPoolExecutor(
  14.                 300,//线程池内核心线程数
  15.                 500,//线程池内最大线程数
  16.                 30,//空闲线程存活时间
  17.                 TimeUnit.SECONDS,
  18.                 new LinkedBlockingDeque<>(50)
  19.         );
  20.     }
  21. }

三.异步代码的具体实现

  1. DimAsyncFunction.java(用于处理异步逻辑)
  1. package com.atguigu.gmall.function;
  3. import com.atguigu.gmall.common.Constant;
  4. import com.atguigu.gmall.util.JdbcUtil;
  5. import com.atguigu.gmall.util.RedisUtil;
  6. import com.atguigu.gmall.util.ThreadPoolUtil;
  7. import org.apache.flink.configuration.Configuration;
  8. import org.apache.flink.streaming.api.functions.async.ResultFuture;
  9. import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
  10. import redis.clients.jedis.Jedis;
  12. import java.sql.Connection;
  13. import java.sql.SQLException;
  14. import java.util.concurrent.ThreadPoolExecutor;
  16. public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> {
  17.     private ThreadPoolExecutor threadPool;
  20.     @Override
  21.     public void open(Configuration parameters) throws Exception {
  22.         threadPool = ThreadPoolUtil.getThreadPool();
  23.     }
  25.     @Override
  26.     public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
  27.         //在线程中去使用客户端去读取维度数据
  28.         threadPool.submit(new Runnable() {
  29.             @Override
  30.             public void run() {
  31.                 Connection phoenixConnection = null;
  32.                 Jedis redisClient = null;
  33.                 try {
  34.                     phoenixConnection = JdbcUtil.getPhoenixConnection(Constant.PHOENIX_URL);
  35.                     redisClient = RedisUtil.getRedisClient();
  36.                     addDim(phoenixConnection,redisClient,input,resultFuture);
  37.                 } catch (SQLException throwables) {
  38.                     throwables.printStackTrace();
  39.                 } catch (ClassNotFoundException e) {
  40.                     e.printStackTrace();
  41.                 } finally {
  42.                     if(phoenixConnection != null){
  43.                         try {
  44.                             phoenixConnection.close();
  45.                         } catch (SQLException throwables) {
  46.                             throwables.printStackTrace();
  47.                         }
  48.                     }
  49.                     if(redisClient != null){
  50.                         redisClient.close();
  51.                     }
  52.                 }
  54.             }
  55.         });
  57.     }
  59.     protected abstract void addDim(Connection phoenixConnection, 
  60.                                    Jedis redisClient, 
  61.                                    T input, 
  62.                                    ResultFuture<T> resultFuture);
  63. }
  1. 具体实现代码
  1. private SingleOutputStreamOperator<OrderWide> joinDims(SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims) {
  2.         //join 6 张维度表
  3.         //根据维度表中的某个id去查找对应的那一条数据
  4.        return AsyncDataStream
  5.                .unorderedWait(
  6.                        orderWideStreamWithoutDims,
  7.                        new DimAsyncFunction<OrderWide>(){
  8.                            @Override
  9.                            protected void addDim(Connection phoenixConnection,
  10.                                                  Jedis redisClient,
  11.                                                  OrderWide orderWide,
  12.                                                  ResultFuture<OrderWide> resultFuture) {
  13.                                //1.补充user
  14.                                JSONObject userInfo = DimUtil.getDim(
  15.                                        phoenixConnection,
  16.                                        redisClient,
  17.                                        "dim_user_info",
  18.                                        orderWide.getUser_id());
  19.                                orderWide.setUser_gender(userInfo.getString("GENDER"));
  20.                                try {
  21.                                    orderWide.calcUserAge(userInfo.getString("BIRTHDAY"));
  22.                                } catch (ParseException e) {
  23.                                    e.printStackTrace();
  24.                                }
  26.                                // 2. 补齐省份
  27.                                JSONObject provinceInfo = DimUtil.getDim(
  28.                                        phoenixConnection,
  29.                                        redisClient,
  30.                                        "dim_base_province",
  31.                                        orderWide.getProvince_id());
  32.                                orderWide.setProvince_name(provinceInfo.getString("NAME"));
  33.                                orderWide.setProvince_iso_code(provinceInfo.getString("ISO_CODE"));
  34.                                orderWide.setProvince_area_code(provinceInfo.getString("AREA_CODE"));
  35.                                orderWide.setProvince_3166_2_code(provinceInfo.getString("ISO_3166_2"));
  37.                                // 3. sku
  38.                                JSONObject skuInfo = DimUtil.getDim(
  39.                                        phoenixConnection,
  40.                                        redisClient,
  41.                                        "dim_sku_info",
  42.                                        orderWide.getSku_id());
  43.                                orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
  45.                                orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
  46.                                orderWide.setTm_id(skuInfo.getLong("TM_ID"));
  47.                                orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
  49.                                // 4. spu
  50.                                JSONObject spuInfo = DimUtil.getDim(
  51.                                        phoenixConnection,
  52.                                        redisClient,
  53.                                        "dim_spu_info",
  54.                                        orderWide.getSpu_id());
  55.                                orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));
  58.                                // 5. tm
  59.                                JSONObject tmInfo =DimUtil.getDim(
  60.                                        phoenixConnection,
  61.                                        redisClient,
  62.                                        "dim_base_trademark",
  63.                                        orderWide.getTm_id());
  64.                                orderWide.setTm_name(tmInfo.getString("TM_NAME"));
  66.                                // 5. tm
  67.                                JSONObject c3Info =DimUtil.getDim(
  68.                                        phoenixConnection,
  69.                                        redisClient,
  70.                                        "dim_base_category3",
  71.                                        orderWide.getCategory3_id());
  72.                                orderWide.setCategory3_name(c3Info.getString("NAME"));
  73.                                resultFuture.complete(Collections.singletonList(orderWide));
  75.                            }
  76.                        },
  77.                        60,
  78.                        TimeUnit.SECONDS
  79.                );
  80.     }
  1. 启动APP测试,发现数据已经在异步处理了

Flink实时数仓(二) - 图11

四.数据写入到kafka

  1.  private void write2Kafka(SingleOutputStreamOperator<OrderWide> stream) {
  2.         stream.map(
  3.                 new MapFunction<OrderWide, String>() {
  4.                     @Override
  5.                     public String map(OrderWide value) throws Exception {
  6.                         return JSON.toJSONString(value);
  7.                     }
  8.                 }
  9.         )
  10.                 .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWM_ORDER_WIDE));
  11.     }

五.将项目打包到集群测试

Flink实时数仓(二) - 图12

发现异步超时,原因是phoenix打包时依赖问题

解决方法是打包时不打包phoenix依赖,在集群的flink-yarn的lib中添加phoenix依赖

  1. /opt/module/phoenix-5.0.0 » cp phoenix-5.0.0-HBase-2.0-client.jar /opt/module/flink-yarn/lib

重启flink-yarn-session

  1. /opt/module/flink-yarn » bin/yarn-session.sh -d

重新启动应用

Flink实时数仓(二) - 图13

发现正常,没有报异步超时,kafka中也有消费到的数据

Flink实时数仓(二) - 图14

第五章.DWM层:支付宽表

1.需求分析

  1. 支付宽表的目的,最主要的原因是支付表没有订单明细,支付金额没有细分到商品上,没有办法统计商品级别的支付情况
  3. 所以本次宽表的核心是把支付表的信息与订单明细关联上
  5. 解决方案有三个
  1. 一个是把订单明细表输出到hbase上,在支付宽表计算时查询hbase,这相当于把订单明细作为一种维度进行管理
  2. 一个是用流的方式接收订单明细,然后用双流join的方式进行合并,因为订单与支付产生有一定的时差,所以必须用intervalJoin来管理流的状态时间,保证当支付到达时订单清洗还保存在状态中
  3. 使用流的方式让支付表与订单宽表进行join,就省去了查询维度表的步骤

2.具体实现

用到的pojo类

  1. PaymentInfo.java
  1. package com.atguigu.gmall.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. import java.math.BigDecimal;
  9. @Data
  10. @AllArgsConstructor
  11. @NoArgsConstructor
  12. public class PaymentInfo {
  13.     private Long id;
  14.     private Long order_id;
  15.     private Long user_id;
  16.     private BigDecimal total_amount;
  17.     private String subject;
  18.     private String payment_type;
  19.     private String create_time;
  20.     private String callback_time;
  21. }
  1. PaymentWide.java
  1. package com.atguigu.gmall.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6. import org.apache.commons.beanutils.BeanUtils;
  8. import java.lang.reflect.InvocationTargetException;
  9. import java.math.BigDecimal;
  11. @Data
  12. @AllArgsConstructor
  13. @NoArgsConstructor
  14. public class PaymentWide {
  16.     private Long payment_id;
  17.     private String subject;
  18.     private String payment_type;
  19.     private String payment_create_time;
  20.     private String callback_time;
  21.     private Long detail_id;
  22.     private Long order_id;
  23.     private Long sku_id;
  24.     private BigDecimal order_price;
  25.     private Long sku_num;
  26.     private String sku_name;
  27.     private Long province_id;
  28.     private String order_status;
  29.     private Long user_id;
  30.     private BigDecimal total_amount;
  31.     private BigDecimal activity_reduce_amount;
  32.     private BigDecimal coupon_reduce_amount;
  33.     private BigDecimal original_total_amount;
  34.     private BigDecimal feight_fee;
  35.     private BigDecimal split_feight_fee;
  36.     private BigDecimal split_activity_amount;
  37.     private BigDecimal split_coupon_amount;
  38.     private BigDecimal split_total_amount;
  39.     private String order_create_time;
  41.     private String province_name;//查询维表得到
  42.     private String province_area_code;
  43.     private String province_iso_code;
  44.     private String province_3166_2_code;
  45.     private Integer user_age;
  46.     private String user_gender;
  48.     private Long spu_id;     //作为维度数据 要关联进来
  49.     private Long tm_id;
  50.     private Long category3_id;
  51.     private String spu_name;
  52.     private String tm_name;
  53.     private String category3_name;
  55.     public PaymentWide(PaymentInfo paymentInfo, OrderWide orderWide) {
  56.         mergeOrderWide(orderWide);
  57.         mergePaymentInfo(paymentInfo);
  59.     }
  61.     public void mergePaymentInfo(PaymentInfo paymentInfo) {
  62.         if (paymentInfo != null) {
  63.             try {
  64.                 BeanUtils.copyProperties(this, paymentInfo);
  65.                 payment_create_time = paymentInfo.getCreate_time();
  66.                 payment_id = paymentInfo.getId();
  67.             } catch (IllegalAccessException | InvocationTargetException e) {
  68.                 e.printStackTrace();
  69.             }
  70.         }
  71.     }
  73.     public void mergeOrderWide(OrderWide orderWide) {
  74.         if (orderWide != null) {
  75.             try {
  76.                 BeanUtils.copyProperties(this, orderWide);
  77.                 order_create_time = orderWide.getCreate_time();
  78.             } catch (IllegalAccessException | InvocationTargetException e) {
  79.                 e.printStackTrace();
  80.             }
  81.         }
  82.     }
  83. }

实现代码

  1. package com.atguigu.gmall.app.dwm;
  3. import com.alibaba.fastjson.JSON;
  4. import com.atguigu.gmall.app.BaseAppV2;
  5. import com.atguigu.gmall.common.Constant;
  6. import com.atguigu.gmall.pojo.OrderWide;
  7. import com.atguigu.gmall.pojo.PaymentInfo;
  8. import com.atguigu.gmall.pojo.PaymentWide;
  9. import com.atguigu.gmall.util.FlinkSinkUtil;
  10. import com.atguigu.gmall.util.IteratorToListUtil;
  11. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  12. import org.apache.flink.api.common.functions.MapFunction;
  13. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  14. import org.apache.flink.streaming.api.datastream.KeyedStream;
  15. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  16. import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
  17. import org.apache.flink.streaming.api.windowing.time.Time;
  18. import org.apache.flink.util.Collector;
  20. import java.time.Duration;
  21. import java.util.HashMap;
  23. public class DwmPaymentWideApp extends BaseAppV2 {
  24.     public static void main(String[] args) {
  25.         new DwmPaymentWideApp().init(3006,1,"DwmPaymentWideApp",
  26.                 "DwmPaymentWideApp", Constant.TOPIC_DWD_PAYMENT_INFO,Constant.TOPIC_DWM_ORDER_WIDE);
  27.     }
  28.     @Override
  29.     public void run(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streams) {
  30.         //streams.get(Constant.TOPIC_DWD_PAYMENT_INFO).print(Constant.TOPIC_DWD_PAYMENT_INFO);
  31.         //streams.get(Constant.TOPIC_DWM_ORDER_WIDE).print(Constant.TOPIC_DWM_ORDER_WIDE);
  32.         KeyedStream<PaymentInfo, Long> paymentInfoStream = streams.get(Constant.TOPIC_DWD_PAYMENT_INFO)
  33.                 .map(new MapFunction<String, PaymentInfo>() {
  34.                     @Override
  35.                     public PaymentInfo map(String value) throws Exception {
  36.                         return JSON.parseObject(value, PaymentInfo.class);
  37.                     }
  38.                 })
  39.                 .assignTimestampsAndWatermarks(
  40.                         WatermarkStrategy.<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  41.                                 .withTimestampAssigner((info, ts) -> IteratorToListUtil.toTs(info.getCreate_time()))
  42.                 )
  43.                 .keyBy(info -> info.getOrder_id());
  45.         KeyedStream<OrderWide, Long> orderWideStream = streams.get(Constant.TOPIC_DWM_ORDER_WIDE)
  46.                 .map(new MapFunction<String, OrderWide>() {
  47.                     @Override
  48.                     public OrderWide map(String value) throws Exception {
  49.                         return JSON.parseObject(value, OrderWide.class);
  50.                     }
  51.                 })
  52.                 .assignTimestampsAndWatermarks(
  53.                         WatermarkStrategy.<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  54.                                 .withTimestampAssigner((wide, ts) -> IteratorToListUtil.toTs(wide.getCreate_time()))
  55.                 )
  56.                 .keyBy(wide -> wide.getOrder_id());
  58.         paymentInfoStream
  59.                 .intervalJoin(orderWideStream)
  60.                 .between(Time.minutes(-45),Time.minutes(10))
  61.                 .process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
  62.                     @Override
  63.                     public void processElement(PaymentInfo left, OrderWide right, Context ctx, Collector<PaymentWide> out) throws Exception {
  64.                         out.collect(new PaymentWide(left,right));
  65.                     }
  66.                 })
  67.                 .map(new MapFunction<PaymentWide, String>() {
  68.                     @Override
  69.                     public String map(PaymentWide value) throws Exception {
  70.                         return JSON.toJSONString(value);
  71.                     }
  72.                 })
  73.                 .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWM_PAYMENT_WIDE));
  74.                 //.print();
  77.     }
  78. }

3.将项目打包到集群上进行测试

Flink实时数仓(二) - 图15

发现数据已经成功采集到kafka