Flink 实时数仓（四）【DWD 层搭建（二）流量域事实表】 ...

前言

        昨天刚搬到新校区，新校区小的可怜，幸亏之后出去实习交通可以方便点；待在学院太受限了，早点离开！
        本日开始完成 DWD 层剩余的需求，上一节我们把日志数据根据不同类型分流写入到了不同的主题；
1、流量域独立访客变乱事实表

        独立访客指的实在就是我们 web 端日志分析指标中常说的 UV，上一节我们已经把页面日志写入到 dwd_page_traffic_log 主题当中了，以是这里我们直接对这个主题进行消耗处理；
1.1、实现思绪

        既然是独立访客，就必须对日志中的数据做去重（独立访客数一般用来做日活指标，因为我们的机器一般都是 24 小时全年无休的，以是我们实时数仓也可以做这种日级别的指标需求，通过状态来存储历史就可以实现），而怎么判断访客是否重复？这就又用到了 Flink 中的状态编程（状态就是历史）；和上一节我们判断新老访客一样，我们这里也可以给每个 mid 维护一个名为 lastVisitDate 的 ValueState（对 mid 进行 keyby），存储上一次访问的日期（留意是日期，只准确到天），每来一条数据就判断它的 lastVisitDate：

• 如果 lastVisitDate 为 null 或者 不是本日，则保留数据，否则抛弃
  

一旦进入第二天，lastVisitDate 状态就应该被清空（设置状态 TTL 为 1 天）
此外，对于 0 点的数据我们这里需要明白统计规则：

• 独立访客数据对应的页面必然是会话起始页面，last_page_id 必为 null；以是对于跨天的访问不能盘算在内（昨天到本日访问了多个页面，而本日页面的 last_page_id 必然不为 null），我们需要在消耗数据后的第一步就需要进行过滤；
  

1.2、代码实现 

1. public class DwdTrafficUniqueVisitorDetail {
3.     public static void main(String[] args) throws Exception {
4.         // TODO 1. 获取执行环境
5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
6.         env.setParallelism(1); // 生产环境中设置为kafka主题的分区数
8.         // 1.1 开启checkpoint
9.         env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
10.         env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/s/ck");
11.         env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
12.         env.getCheckpointConfig().setMaxConcurrentCheckpoints(2); // 设置最大共存的checkpoint数量
13.         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L)); // 固定频率重启: 尝试3次重启,每5s重启一次
15.         // 1.2 设置状态后端
16.         env.setStateBackend(new HashMapStateBackend());
18.         // TODO 2. 消费 kafka dwd_traffic_page_log 主题
19.         String topic = "dwd_traffic_page_log";
20.         String groupId = "uvDetail";
21.         DataStreamSource<String> pageDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
23.         // TODO 3. 过滤 last_page_id != Null 的数据
24.         // 使用 flatMap 而没用 filter,因为 flatMap 可以把过滤和转json 两步都一起完成
25.         SingleOutputStreamOperator<JSONObject> jsonDS = pageDS.flatMap(new FlatMapFunction<String, JSONObject>() {
26.             @Override
27.             public void flatMap(String value, Collector<JSONObject> out) throws Exception {
28.                 try {
29.                     JSONObject jsonObject = JSONObject.parseObject(value);
30.                     // 获取 last_page_id
31.                     String last_page_id = jsonObject.getJSONObject("page").getString("last_page_id");
32.                     if (last_page_id == null) {
33.                         out.collect(jsonObject);
34.                     }
35.                 } catch (Exception e) {
36.                     e.printStackTrace();
37.                 }
38.             }
39.         });
41.         // TODO 4. 按照 mid 分组
42.         KeyedStream<JSONObject, String> keyedStream = jsonDS.keyBy(json -> json.getJSONObject("common").getString("mid"));
44.         // TODO 5. 使用状态编程实现按照 mid 的日期进行去重
45.         // 使用富函数,因为富函数提供更多的信息如上下文等
46.         SingleOutputStreamOperator<JSONObject> uvDS = keyedStream.filter(new RichFilterFunction<JSONObject>() {
47.             private ValueState<String> lastVisitDate = null;
49.             @Override
50.             public void open(Configuration parameters) throws Exception {
51.                 ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("lastVisit", String.class);
52.                 lastVisitDate = getRuntimeContext().getState(stateDescriptor);
53.             }
55.             @Override
56.             public boolean filter(JSONObject value) throws Exception {
57.                 // 获取状态数据 & 当前数据中的时间并转为日期
58.                 String lastDate = lastVisitDate.value();
59.                 Long ts = value.getLong("ts");
60.                 String curDate = DateFormatUtil.toDate(ts);
61.                 if (lastDate == null || !lastDate.equals(curDate)) {
62.                     // 更新状态
63.                     lastVisitDate.update(curDate);
64.                     return true;
65.                 }
66.                 return false;
67.             }
68.         });
70.         // TODO 6. 数据写入 kafka
71.         String targetTopic = "dwd_traffic_unique_visitor_detail";
72.         uvDS.map(data -> data.toJSONString()).addSink(MyKafkaUtil.getFlinkKafkaProducer(targetTopic));
74.         // TODO 7. 执行任务
75.         env.execute("DwdTrafficUniqueVisitorDetail");
76.     }
78. }

复制代码

1.3、TTL 优化

        上面我们的代码逻辑看起来已经没什么问题了，但是我们可以设想：假设一个用户,2024-01-01 初次登录之后，它的 lastVisitDate 状态会不停存储 2024-01-01，如果他下一次登录是在 2024-12-31，那么期间的 364 天我们依然要不停存储它的状态；而我们判断用户是否已经登录的逻辑是：lastVisitDate 是否为null 或者 lastVisitDate<本日，以是我们完全可以在一天之后把该用户的 lastVisitDate 状态清空，来淘汰状态的生存开销！
TTL 是给状态描述器设置的，而状态描述器是构造状态对象的必须参数！
TTL 是状态的一个属性，当我们修改状态值的时候，TTL 本身并不会更新！这里，我们需要在状态描述器中设置 TTL 的更新计谋为创建或更新状态值的时候就更新 TTL ，重新开始逾期倒计时；
我们只需要修改上面第 5 步，在初始化状态时，在状态描述器中给状态添加 TTL 属性：

1.             @Override
2.             public void open(Configuration parameters) throws Exception {
3.                 ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("lastVisit", String.class);
4.                 // 给状态添加 TTL
5.                 stateDescriptor.enableTimeToLive(new StateTtlConfig.Builder(Time.days(1))
6.                         // 设置 TTL 可更新,并且在创建或更新状态的时候更新
7.                         .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
8.                         .build()
9.                 );
10.                 lastVisitDate = getRuntimeContext().getState(stateDescriptor);
11.             }

复制代码

2、流量域用户跳失变乱事实表

        跳出的概念：跳出指的是用户在一次会话中只访问了一个页面的环境（留意：粒度是会话），我们在之前做离线数仓的时候做过跳出率的指标，对于离线数仓，我们可以在 DWS 层构建一张流量域近1日会话粒度页面浏览表（dws_traffic_session_page_view_1d），通过下面的 SQL 就可以统计出该指标：

1. SELECT
2. CAST(SUM(IF(page_count=1,1,0))/COUNT(*)) AS DECIMAL(16,2) AS bounce_rate
3. FROM dws_traffic_session_page_view_1d

复制代码

        在这里的实时数仓中，我们不大概等到一天结束最后才去盘算跳出率；但是我们这里又没有 session_id，以是我们只能换一种思绪：
思绪1（会话窗口）：

• 利用会话窗口，为每个 mid 开启一个会话窗口并指定间隔为 10 s；一旦到了 10s 触发窗口关闭，盘算窗口内的数据条数，> 1 条则说明这次会话没有发生跳出；
  

        这种思绪的问题很显着：① 如果我短时间（10s内）发生多个跳出，但是恰好这些跳出都在一个会话，这会导致窗口结束时误以为这不是跳出，毕竟窗口内有多条数据；② 大概我的一次正常的会话，被会话窗口切分到两个不同的会话窗口，效果把一个非跳出访问盘算为 2 个跳出访问；
思绪2（状态编程）：
        在离线数仓中，当我们没有 session_id 时，我们可以一天的数据按照 mid 进行分组，然后根据时间戳字段进行排序，这样来盘算一个 session；但是这里是实时数仓，我们不知道什么时候一个 session 会结束，以是我们可以设置一个定时器，定时器时间范围内的数据如果没数据来就视作一个会话结束，触发盘算；并结合状态编程，把新会话的首页存入状态

• 遇到 last_page 为 null 的数据就试着取出状态
  
  
  
  • 如果状态为 null，则该页面是新的会话起始页，开启定时器并将数据自身写入状态
    
  • 如果状态不为 null，说明刚跳出一次，并且在定时器时间范围内又进来一次；这种环境需要将第一条数据（跳出的数据，也就是写入状态中的数据）输出，然后将自身写入状态，定时器依然存在，等时间到了触发盘算
    
  
  
• 如果 last_page 不为 null，则状态中的数据和该条数据都抛弃
  

这种思绪同样存在问题，当数据是乱序的时候一切都就乱套了；
思绪3（Flink CEP）：
Flink CEP 实在就是利用 状态编程 + within 开窗 来处理这种复杂变乱

Flink CEP 界说的规则之间的连续计谋

• 严格连续: 盼望所有匹配的变乱严格的一个接一个出现，中间没有任何不匹配的变乱。对应方法为 next()；
  
• 疏松连续: 忽略匹配的变乱之间的不匹配的变乱。对应方法为followedBy()；
  
• 不确定的疏松连续: 更进一步的疏松连续，允许忽略掉一些匹配变乱的附加匹配。对应方法为followedByAny()。
  

界说模式之前的代码
        这里需要留意：因为我们反面要保证数据有序，以是我们最好指定变乱时间的提取字段，并添加水位线设置公道的超时时间（理论上可以保证数据绝对有序）：

1. public class DwdTrafficUserJumpDetail {
3.     public static void main(String[] args) {
4.         // TODO 1. 获取执行环境
5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
6.         env.setParallelism(1); // 生产环境中设置为kafka主题的分区数
8.         // 1.1 开启checkpoint
9.         env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
10.         env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/s/ck");
11.         env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
12.         env.getCheckpointConfig().setMaxConcurrentCheckpoints(2); // 设置最大共存的checkpoint数量
13.         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L)); // 固定频率重启: 尝试3次重启,每5s重启一次
15.         // 1.2 设置状态后端
16.         env.setStateBackend(new HashMapStateBackend());
18.         // TODO 2. 消费 kafka dwd_traffic_page_log 主题
19.         String topic = "dwd_traffic_page_log";
20.         String groupId = "user_jump_detail";
21.         DataStreamSource<String> pageDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
23.         // TODO 3. 将数据转为 JSON
24.         SingleOutputStreamOperator<JSONObject> jsonDS = pageDS.map(JSON::parseObject);
26.         // TODO 4. 提取事件时间 & 按照 mid 分组
27.         KeyedStream<JSONObject, String> keyedStream = jsonDS
28.                 .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
29.                     .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
30.                         @Override
31.                         public long extractTimestamp(JSONObject element, long recordTimestamp) {
32.                             return element.getLong("ts");
33.                         }
34.                     }))
35.                 .keyBy(json -> json.getJSONObject("common").getString("mid"));

复制代码

接下来是核心的界说 CEP 模式的代码：

1.         // TODO 5. 定义 CEP 模式序列
2.         // 泛型方法类型指的是流的类型(下面的 start 和 next 作为提取事件的 key)
3.         Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
4.             @Override
5.             public boolean filter(JSONObject value) throws Exception {
6.                 return value.getJSONObject("page").getString("last_page_id") == null;
7.             }
8.         }).next("next").where(new SimpleCondition<JSONObject>() {
9.             @Override
10.             public boolean filter(JSONObject value) throws Exception {
11.                 return value.getJSONObject("page").getString("last_page_id") == null;
12.             }
13.         }).within(Time.seconds(10L));
15.         // 等价于 循环模式 共用一个 key: start
16.         Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
17.             @Override
18.             public boolean filter(JSONObject value) throws Exception {
19.                 return value.getJSONObject("page").getString("last_page_id") == null;
20.             }
21.         })
22.                 .times(2) // 默认是宽松近邻 followedBy
23.                 .consecutive() // 严格近邻 next
24.                 .within(Time.seconds(10L));
26.         // TODO 6. 建模式序列作用到流上
27.         PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);
29.         // TODO 7. 提取事件(匹配上的时间 和 超时时间)
30.         OutputTag<String> timeoutTag = new OutputTag<>("timeout");
31.         SingleOutputStreamOperator<String> selectDS = patternStream.select(timeoutTag,
32.                 // 超时数据
33.                 new PatternTimeoutFunction<JSONObject, String>() {
34.                     // 对于超时数据来说,当前的数据第一个规则匹配上了,第二个没有匹配上导致超时,那么我们要提取的就是当前数据(第一个数据,第二个数据没来)
35.                     // 这里的 Map 的 v 是 List 数据类型,因为考虑到我们可能使用的是循环模式(只有一个key)
36.                     @Override
37.                     public String timeout(Map<String, List<JSONObject>> map, long l) throws Exception {
38.                         return map.get("start").get(0).toJSONString();
39.                     }
40.                 },
41.                     // 匹配上的数据
42.                     new PatternSelectFunction<JSONObject, String>() {
43.                     // 匹配上的数据,我们只要第一个数据,因为只能证明第一个数据是跳出数据
44.                     @Override
45.                     public String select(Map<String, List<JSONObject>> map) throws Exception {
46.                         return map.get("start").get(0).toJSONString();
47.                     }
48.                 });
49.         DataStream<String> timeoutDS = selectDS.getSideOutput(timeoutTag);
51.         // TODO 8. 合并两种事件
52.         DataStream<String> unionDS = selectDS.union(timeoutDS);
54.         // TODO 9. 合并后的数据写入 kafka
55.         String targetTopic = "dwd_traffic_user_jump_detail";
56.         unionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(targetTopic));
58.         // TODO 10. 启动任务
59.         env.execute("DwdTrafficUserJumpDetail");
60.     }
61. }

复制代码

上面我们界说了两种匹配规则：

• 第一条数据的 last_page_id 为 null ，且超时没有收到第二条数据，认定该条数据为跳出数据
  
• 第二条数据的 last_page_id 为 null ，则认定第一条数据是跳出数据
  

        超时时间内规则一被满意，未等到第二条数据则会被判断为超时数据。以是我们只要把超时数据和 满意连续两条数据的 last_page_id 均为 null 中的第一条数据 union 起来，得到的即为答案所需数据；
总结

        至此，流量域的三个需求都已经完成；

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。