03Flink实时数仓（尚硅谷）- DWM 层业务实现

本博客内容出自尚硅谷b站公开课，如有侵权，请联系博主删除

第1章 DWS 层与 DWM 层的设计

1.1 设计思路

我们在之前通过分流等手段，把数据分拆成了独立的 Kafka Topic。那么接下来如何处理数据，就要思考一下我们到底要通过实时计算出哪些指标项。

因为实时计算与离线不同，实时计算的开发和运维成本都是非常高的，要结合实际情况考虑是否有必要象离线数仓一样，建一个大而全的中间层。

如果没有必要大而全，这时候就需要大体规划一下要实时计算出的指标需求了。把这些指标以主题宽表的形式输出就是我们的 DWS 层。

1.2 需求梳理

当然实际需求还会有更多，这里主要以可视化大屏为目的进行实时计算的处理。

DWM 层的定位是什么，DWM 层主要服务 DWS，因为部分需求直接从 DWD 层到 DWS 层中间会有一定的计算量，而且这部分计算的结果很有可能被多个 DWS 层主题复用，所以部分 DWD 成会形成一层 DWM，我们这里主要涉及业务。

• 访问 UV 计算
• 跳出明细计算
• 订单宽表
• 支付宽表

第2章 DWM 层-访客 UV 计算

2.1 需求分析与思路

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

那么如何从用户行为日志中识别出当日的访客，那么有两点：

• 其一，是识别出该访客打开的第一个页面，表示这个访客开始进入我们的应用
• 其二，由于访客可以在一天中多次进入应用，所以我们要在一天的范围内进行去重

2.2 代码实现

实现思路：

• 首先用 keyby 按照 uid 进行分组
• 分组后使用 keystate 状态，记录用户进入时间，实现 RichFilterFunction 完成过滤
• 重写 open 方法用来初始化状态
• 重写 filter 方法进行过滤
  • 可以直接筛掉 last_page_id 不为空的字段，因为只要有上一页，说明这条不是这个用户进入的首个页面。
  • 状态用来记录用户的进入时间，只要这个 lastVisitDate 是今天，就说明用户今天已经访问过了所以筛除掉。如果为空或者不是今天，说明今天还没访问过，则保留。
  • 因为状态值主要用于筛选是否今天来过，所以这个记录过了今天基本上没有用了，这里 enableTimeToLive 设定了 1 天的过期时间，避免状态过大。

//4.按照 mid 分组
KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("user").getString("uid"));

//5.过滤掉不是今天第一次访问的数据
SingleOutputStreamOperator<