Flink 实时数仓（八）【DWS 层搭建（二）流量域、用户域、交易域搭建】

前言

今天的任务是完成流量域最后一个需求、用户域的两个需求以及交易域的部分需求；

1、流量域页面浏览各窗口汇总表

任务：从 Kafka 页面日志主题读取数据，统计当日的首页和商品详情页独立访客数。

注意：一般我们谈到访客，指的是 mid；而用户才是 uid；

1.1、思路

消费 Kafka dwd_traffic_page_log
过滤出 page_id = 'home' 或 'good_detail' 的数据
按照 mid 分组
使用状态编程为每个 mid 维护两个状态：首页的末次访问日期，商品详情页的末次访问日期
- 每新来一条数据就判断它的两个状态是否为 null
  - 如果为 null，则给状态赋值
  - 如果不为 null，则不做操作
- 当两个状态中有一个不为 null 时，发送数据到下游
开窗聚合（实时计算更新报表，这里开窗用的是 windowAll() ，因为上一步发送下来的数据已经不再是键控流了）
写出到 clickhouse

1.2、实现

1.2.1、创建 ck 表并创建 Java Bean

首先创建 ck 表结构，和前面的表一样，主要的字段就是：维度 + 度量值（这里没有粒度，因为我们统计的是一个宏观的统计结果信息，到 ADS 都不用加工），这里的 stt 和 edt 依然是作为 ck 表的 order by 字段防止数据重复；ts 字段作为 ck 的版本字段；这里 order by 字段取窗口起止时间，因为窗口是基于事件时间的，所以不用担心任务挂了之后重复消费造成数据重复的问题，ck 会自动根据 order by 字段进行去重；

create table if not exists dws_traffic_page_view_window
(stt               DateTime,edt               DateTime,home_uv_ct        UInt64,good_detail_uv_ct UInt64,ts                UInt64
) engine = ReplacingMergeTree(ts)partition by toYYYYMMDD(stt)order by (stt, edt);

创建 ck 表对应的 JavaBean：

@Data
@AllArgsConstructor
public class TrafficHomeDetailPageViewBean {// 窗口起始时间String stt;// 窗口结束时间String edt;// 首页独立访客数Long homeUvCt;// 商品详情页独立访客数Long goodDetailUvCt;// 时间戳Long ts;
}

1.2.2、读取页面日志并过滤出首页与商品详情页

这里我们不仅要过滤还希望尽量顺便把数据转换为 JSONObject 格式，所以选用 flatMap 最为合适：

过滤出 page_id 为 home 或者 good_detail 的数据

// TODO 3. 读取 dwd_traffic_page_log 的数据String groupId = "dws_traffic_page_view_window";DataStreamSource<String> pageLog = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log", groupId));// TODO 4. 转为 json 并过滤出首页和商品详情页SingleOutputStreamOperator<JSONObject> filterDS = pageLog.flatMap(new FlatMapFunction<String, JSONObject>() {@Overridepublic void flatMap(String value, Collector<JSONObject> out) throws Exception {JSONObject jsonObject = JSON.parseObject(value);String page_id = jsonObject.getJSONObject("page").getString("page_id");if (page_id != null) {if (page_id.equals("home") || page_id.equals("good_detail")) {out.collect(jsonObject);}}}});

1.2.3、提取事件时间并生成水位线

// TODO 5. 提取事件时间生成水位线filterDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {@Overridepublic long extractTimestamp(JSONObject element, long recordTimestamp) {return element.getLong("ts");}}));

1.2.4、状态编程过滤出独立访客

这里使用富函数的 flatMap，因为富函数中才有 open（在 open 方法中初始化状态）、close等方法，以及获取上下文对象（通过上下文对象给状态描述器设置ttl并初始化）等高级操作；

这里 flatMap 的输出类型我们设置为之前写好的 ck 表对应的 JavaBean ，方便直接插入到 ck中；

这里我们同样可以给状态设置一个 TTL 防止长时间访客未访问状态存储浪费；这里两个状态任意一个不为 null 即可输出：

        // TODO 6. 状态编程（按照mid分组）过滤出独立访客KeyedStream<JSONObject, String> keyedStream = filterDS.keyBy(json -> json.getJSONObject("common").getString("mid"));SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> trafficHomeDetailDS = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, TrafficHomeDetailPageViewBean>() {private ValueState<String> homeLastVisit;private ValueState<String> detailLastVisit;@Overridepublic void open(Configuration parameters) throws Exception {StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build();ValueStateDescriptor<String> homeStateDescriptor = new ValueStateDescriptor<>("home-state", String.class);ValueStateDescriptor<String> detailStateDescriptor = new ValueStateDescriptor<>("detail-state", String.class);// 设置 TTLhomeStateDescriptor.enableTimeToLive(ttlConfig);detailStateDescriptor.enableTimeToLive(ttlConfig);homeLastVisit = getRuntimeContext().getState(homeStateDescriptor);detailLastVisit = getRuntimeContext().getState(detailStateDescriptor);}@Overridepublic void flatMap(JSONObject value, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {// 获取状态数据以及当前数据中的日期String curDt = DateFormatUtil.toDate(value.getLong("ts"));String homeLastDt = homeLastVisit.value();String detailLastDt = detailLastVisit.value();long homeUvCt = 0;long goodDetailUvCt = 0;if (homeLastDt == null || !homeLastDt.equals(curDt)) {homeUvCt = 1;homeLastVisit.update(curDt);}if (detailLastDt == null || !detailLastDt.equals(curDt)) {goodDetailUvCt = 1;detailLastVisit.update(curDt);}if (homeUvCt == 1 || goodDetailUvCt == 1) {out.collect(new TrafficHomeDetailPageViewBean("", "",homeUvCt,goodDetailUvCt,value.getLong("ts")));}}});

1.2.5、开窗聚合并写入到 clickhouse

这里的窗口函数依旧是先用增量聚合函数，再用全量聚合函数（获得窗口信息）；

注意：这里的 ts 字段是 clickhouse 表数据的版本字段，取系统时间即可；

// TODO 7. 开窗（windowAll聚合）聚合SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> resultDS = trafficHomeDetailDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))).reduce(new ReduceFunction<TrafficHomeDetailPageViewBean>() {@Overridepublic TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean value1, TrafficHomeDetailPageViewBean value2) throws Exception {value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());value1.setGoodDetailUvCt(value1.getGoodDetailUvCt() + value2.getGoodDetailUvCt());return value1;}}, new AllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {@Overridepublic void apply(TimeWindow window, Iterable<TrafficHomeDetailPageViewBean> values, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {TrafficHomeDetailPageViewBean next = values.iterator().next();next.setTs(System.currentTimeMillis());next.setStt(DateFormatUtil.toYmdHms(window.getStart()));next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));out.collect(next);}});// TODO 8. 写入到 clickhouseresultDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_traffic_page_view_window values(?,?,?,?,?)"));// TODO 9. 启动任务env.execute("DwsTrafficPageViewWindow");

2、用户域用户登陆各窗口汇总表

任务：从 Kafka 页面日志主题读取数据，统计七日回流用户和当日独立用户数。

当日独立用户数很好求，和上面差不多，也是使用状态编程对 uid 保存状态去重即可。接下来我们主要分析七日回流用户怎么求：

2.1、思路分析

回流用户定义：之前的活跃用户，一段时间未活跃（流失），今日又活跃了。这里要求统计回流用户总数，规定当日登陆，且自上次登陆之后至少 7 日未登录的用户为回流用户。

1、消费页面浏览主题（dwd_traffic_page_log）登录用户过滤：

用户打开应用自动登录（cookie）
- uid != null && last_page_id = null （后面这个条件可以过滤掉没必要的数据）
用户在登录页登录
- uid != null && last_page_id = login

2、设置水位线、uid 分组之后进行状态编程

判断 lastLoginDt 是否为 null
- null：是今天的独立用户，但不是回流用户
- !=null
  - 判断和今天是否相同
    - 相同：丢弃
    - 不同：是今天的独立用户，再判断今天-lastLoginDt >= 8？是回流用户：不是

2.2、代码实现

2.2.1、创建 ck 表并创建对应 JavaBean

这张表依然没有粒度，直接就是统计结果；我们去重的字段依然是窗口的起止时间：

create table if not exists dws_user_user_login_window
(stt     DateTime,edt     DateTime,	back_ct UInt64,uu_ct   UInt64,ts      UInt64
) engine = ReplacingMergeTree(ts)partition by toYYYYMMDD(stt)order by (stt, edt);

import lombok.AllArgsConstructor;
import lombok.Data;@Data
@AllArgsConstructor
public class UserLoginBean {// 窗口起始时间String stt;// 窗口终止时间String edt;// 回流用户数Long backCt;// 独立用户数Long uuCt;// 时间戳Long ts;
}

2.2.2、消费 dwd_traffic_page_log 主题

// TODO 3. 读取 dwd_traffic_page_log 的数据String groupId = "dws_user_user_login_window";DataStreamSource<String> pageLog = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log", groupId));

2.2.3、转换数据流为 JSON 格式并过滤出独立用户

// TODO 4. 转换为 json 格式 & 过滤出独立用户(uid!=null & last_page_id=null 或者 uid!=null & last_page_id=login)SingleOutputStreamOperator<JSONObject> filterDS = pageLog.flatMap(new RichFlatMapFunction<String, JSONObject>() {@Overridepublic void flatMap(String value, Collector<JSONObject> out) throws Exception {JSONObject jsonObject = JSONObject.parseObject(value);String uid = jsonObject.getJSONObject("common").getString("uid");String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");if (uid != null) {if (lastPageId == null || lastPageId.equals("login")) {out.collect(jsonObject);}}}});

2.2.4、提取事件时间生成水位线

        // TODO 5. 提取事件时间生成水位线filterDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {@Overridepublic long extractTimestamp(JSONObject element, long recordTimestamp) {return element.getLong("ts");}}));

2.2.5、使用状态编程过滤出独立用户

// TODO 6. 状态编程过滤出独立用户KeyedStream<JSONObject, String> keyedStream = filterDS.keyBy(json -> json.getJSONObject("common").getString("uid"));SingleOutputStreamOperator<UserLoginBean> userLoginDS = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, UserLoginBean>() {private ValueState<String> lastLoginDtState;@Overridepublic void open(Configuration parameters) throws Exception {StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(7)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build();ValueStateDescriptor<String> lastLoginStateDescriptor = new ValueStateDescriptor<String>("last-login", String.class);lastLoginStateDescriptor.enableTimeToLive(ttlConfig);lastLoginDtState = getIterationRuntimeContext().getState(lastLoginStateDescriptor);}@Overridepublic void flatMap(JSONObject value, Collector<UserLoginBean> out) throws Exception {// 本次登录日期Long curTs = value.getLong("ts");String curDt = DateFormatUtil.toDate(curTs);// 上次登录日期String lastLoginDt = lastLoginDtState.value();long uuCt = 0L;long backCt = 0L;if (lastLoginDt == null) {uuCt = 1;lastLoginDtState.update(curDt);} else if (!lastLoginDt.equals(curDt)) {uuCt = 1;lastLoginDtState.update(curDt);// 判断相差是否 >= 8 天Long lastTs = DateFormatUtil.toTs(lastLoginDt);long days = (curTs - lastTs) / 1000 / 3600 / 24;backCt = days >= 8 ? 1 : 0;}if (uuCt != 0) {out.collect(new UserLoginBean("", "", backCt, uuCt, curTs));}}});

2.2.6、窗口聚合

和上一个需求一样，增量聚合函数和全量聚合函数配合着使用；

// TODO 6. 窗口聚合SingleOutputStreamOperator<UserLoginBean> resultDS = userLoginDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))).reduce((record1, record2) -> {record1.setUuCt(record1.getUuCt() + record2.getUuCt());record2.setBackCt(record1.getBackCt() + record2.getBackCt());return record1;}, new AllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>() {@Overridepublic void apply(TimeWindow window, Iterable<UserLoginBean> values, Collector<UserLoginBean> out) throws Exception {UserLoginBean next = values.iterator().next();next.setStt(DateFormatUtil.toYmdHms(window.getStart()));next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));next.setTs(System.currentTimeMillis());out.collect(next);}});

2.2.7、写出到 clickhouse

// TODO 7. 写入到 clickhouseresultDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_user_user_login_window values(?,?,?,?,?)"));// TODO 8. 启动任务env.execute("DwsUserUserLoginWindow");

3、用户域用户注册各窗口汇总表

任务：从 DWD 层用户注册表中读取数据，统计各窗口注册用户数，写入 ClickHouse。

这个需求比较简单，因为我们之前在 DWD 层已经创建了用户注册事务事实表（包含字段：user_id，date_id，create_time，ts）

3.1、代码实现

这里教程中用的是 DataStream API ，但是我这里想用 Flink SQL 实现：

3.1.1、创建 dwd_user_register 表并生成水位线

注意：当原表中有更贴近事件时间的字段时，我们就尽量少用 Maxwell 的 ts 字段！

// TODO 3. 消费 Kafka dwd_user_register 主题(生成水位线)String groupId = "dws_user_user_register_window";tableEnv.executeSql("CREATE TABLE dwd_user_register " +"`user_id` string," +"`date_id` string," +"`create_time` string," +"`ts` string" +"time_ltz AS TO_TIMESTAMP(FROM_UNIXTIME(create_time/1000)), " +"WATERMARK FOR time_ltz AS time_ltz - INTERVAL '2' SECOND " +")" + MyKafkaUtil.getKafkaDDL("dwd_user_register",groupId));

3.1.2、分组开窗聚合

用 Flink SQL 实现就简单多了，这里的聚合逻辑更简单，直接 count(*)：

// TODO 4. 分组,开窗,聚合Table resultTable = tableEnv.sqlQuery("SELECT " +"    date_format(tumble_start(time_ltz,interval '10' second),'yyyy-MM-dd HH:mm:ss') stt," +"    date_format(tumble_end(time_ltz,interval '10' second),'yyyy-MM-dd HH:mm:ss') edt," +"    count(*) register_ct," +"    unix_timestamp() ts" +"FROM dwd_user_register " +"GROUP BY tumble(time_ltz,interval '10' second)");tableEnv.createTemporaryView("result_table",resultTable);

3.1.3、创建 ck 表及其 Bean

create table if not exists dws_user_user_register_window
(stt         DateTime,edt         DateTime,register_ct UInt64,ts          UInt64
) engine = ReplacingMergeTree(ts)partition by toYYYYMMDD(stt)order by (stt, edt);

这里需要把动态表转为流，所以我们需要创建一个 Java Bean，对应上 ck 表的每个字段：

@Data
@AllArgsConstructor
public class UserRegisterBean {// 窗口起始时间String stt;// 窗口终止时间String edt;// 注册用户数Long registerCt;// 时间戳Long ts;
}

3.1.4、将动态表转为流并写入到 clickhouse

 // TODO 5. 将动态表转为流并写入到 clickhouseDataStream<UserRegisterBean> dataStream = tableEnv.toAppendStream(resultTable, UserRegisterBean.class);dataStream.addSink(ClickHouseUtil.getSinkFunction("insert into dws_user_user_register_window values (?,?,?,?)"));// TODO 6. 启动任务env.execute("DwsUserUserRegisterWindow");

4、交易域加购各窗口汇总表

任务：从 Kafka 读取用户加购明细数据，统计每日各窗口加购独立用户数，写入 ClickHouse。

4.1、思路分析

思路很简单，还是根据 uid 进行 keyby，然后使用状态编程维护一个 lastCartAddDate，对数据进行判断：

如果 lastCartAddDate = null
- 写入状态
如果 lastCartAddDate != null
- 如果 lastCartAddDate != curDate
  - 更新状态
- 否则丢弃

4.2、代码实现

这里不多介绍，和前面的逻辑都是一样的，只说明部分点：

我们在生成水位线的时候，应该尽可能的生成贴近事件时间的，而这里对于加购操作来说，它有两种情况：
- insert：就是加购，会影响的到 create_time 字段
- update：可能是加购，会影响到 operate_time 字段，我们在 DWD 层已经过滤过了：只要 sku_num 变大就是加购
所以这里我们的水位线可以取 operate_time 字段，取不到再取 create_time

// TODO 3. 读取 dwd_traffic_card_add 的数据String groupId = "dws_trade_cart_add_uu_window";DataStreamSource<String> cartAddLog = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_trade_cart_add", groupId));//TODO 4. 转为 json 格式并SingleOutputStreamOperator<JSONObject> jsonDS = cartAddLog.map(JSONObject::parseObject);// TODO 5. 提取事件时间生成水位线jsonDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {@Overridepublic long extractTimestamp(JSONObject element, long recordTimestamp) {String operate_time = element.getString("operate_time");if (operate_time != null){return DateFormatUtil.toTs(operate_time,true);}return DateFormatUtil.toTs(element.getString("create_time"));}}));// TODO 6. 按照用户id进行分组 & 过滤出独立用户KeyedStream<JSONObject, String> keyedStream = jsonDS.keyBy(json -> json.getJSONObject("common").getString("uid"));SingleOutputStreamOperator<CartAddUuBean> filterDS = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, CartAddUuBean>() {private ValueState<String> lastCartAddDateState;@Overridepublic void open(Configuration parameters) throws Exception {StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build();ValueStateDescriptor<String> lastCartAddStateDescriptor = new ValueStateDescriptor<String>("last-cart-add", String.class);lastCartAddStateDescriptor.enableTimeToLive(ttlConfig);lastCartAddDateState = getRuntimeContext().getState(lastCartAddStateDescriptor);}@Overridepublic void flatMap(JSONObject value, Collector<CartAddUuBean> out) throws Exception {// 当前的时间戳Long curTs = value.getLong("ts");String curDate = DateFormatUtil.toDate(curTs);String lastCartAddDate = lastCartAddDateState.value();if (lastCartAddDate == null || !lastCartAddDate.equals(curDate)) {lastCartAddDateState.update(curDate);out.collect(new CartAddUuBean("","",1L,curTs));}}});// TODO 7. 开窗聚合(补充字段)SingleOutputStreamOperator<CartAddUuBean> resultDS = filterDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))).reduce(new ReduceFunction<CartAddUuBean>() {@Overridepublic CartAddUuBean reduce(CartAddUuBean value1, CartAddUuBean value2) throws Exception {value1.setCartAddUuCt(value1.getCartAddUuCt() + value2.getCartAddUuCt());return value1;}}, new AllWindowFunction<CartAddUuBean, CartAddUuBean, TimeWindow>() {@Overridepublic void apply(TimeWindow window, Iterable<CartAddUuBean> values, Collector<CartAddUuBean> out) throws Exception {CartAddUuBean next = values.iterator().next();next.setStt(DateFormatUtil.toYmdHms(window.getStart()));next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));next.setTs(System.currentTimeMillis());out.collect(next);}});// TODO 8. 写出到 clickhouseresultDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_trade_cart_add_uu_window values (?,?,?,?)"));// TODO 9. 启动任务env.execute("DwsTradeCartAddUuWindow");

5、交易域支付各窗口汇总表

任务：从 Kafka 读取交易域支付成功主题数据，统计支付成功独立用户数和首次支付成功用户数（第一次在平台消费）。

5.1、思路分析

如果一个用户是首次支付成功用户（既然是历史第一次下单操作，必然也是今天的第一次下单），那么他必然是今天的支付成功独立用户；所以我们只需要通过状态过滤出 lastPayDate = null 或者 lastPayDate != curDt 的用户（注意：这里的 lastPayDate 不能设置 TTL ，因为我们需要知道这个用户历史上有没有支付过，所以就不允许状态失效）

left join 实现过程：

假设 A 表作为主表与 B 表做等值左外联。当 A 表数据进入算子，而 B 表数据未至时会先生成一条 B 表字段均为 null 的关联数据ab1，其标记为 +I。其后，B 表数据到来，会先将之前的数据撤回，即生成一条与 ab1 内容相同，但标记为 -D 的数据，再生成一条关联后的数据，标记为 +I。这样生成的动态表对应的流称之为回撤流。

在 DWD 层的订单预处理表（dwd_trade_order_pre_process）生成过程中会形成回撤流，因为它需要对订单明细活动表和订单明细优惠券表进行 left join。而我们这里的支付成功依赖于 DWD 层支付成功事务事实表（dwd_trade_pay_detail_suc），该表又依赖于 DWD 层的下单事务事实表（dwd_trade_order_detail），所以这里我们需要考虑回撤流的问题：

回撤数据在 Kafka 中以 null 值的形式存在，只需要简单判断即可过滤。我们需要考虑的是如何对其余数据去重：

order_id = 1001
order_detail_id = 1001-a
order_detail_activity_id: a1SELECT ...
FROM
order_detail od 
join order_info oi
onod.order_id = oi.id
left join order_detail_activity oa
onod.id = oa.order_detail_id

上面我们有一个订单（id=1001），这个订单内只有一个商品并且参与了活动，那么由于 order_detail_activity 来得肯定要晚一些，所以可能会出现下面这种情况：

+/-    order_id    order_detail_id     order_detail_activity_id+        1001        1001-a                    null
-        null        null                      null
+        1001        1001-a                    a1

我们过滤 null 值指的是过滤上面操作是 '-' 的数据，因为回撤数据在 Kafka 中以 null 值的形式存在。而除了 null 值之外，我们还应该过滤掉旧的错误数据，由于 order_detail_activity 数据来得晚一些，导致flink 直接给字段 order_detail_activity_id 一个 null，所以我们应该把这个字段值删除；

但是，对于这个需求（求支付成功的用户数），其实我们也可以不做去重，放到最后再做去重，为什么呢？设想如果一个用户下了多个订单，而我们的支付成功表的粒度是商品，所以数据即使在 left join 之后对相同 order_detail_id 的数据做了去重，但是多个订单的话最终还有重复。

考虑到之后还可能遇到需要去重的需求（尤其是设计到金额的），这里我们还是练习一下如何实现去重：

5.2、代码实现

5.2.1、创建 clickhouse 表格及对应的 JavaBean

create table if not exists dws_trade_payment_suc_window
(stt                           DateTime,edt                           DateTime,payment_suc_unique_user_count UInt64,payment_new_user_count        UInt64,ts                            UInt64
) engine = ReplacingMergeTree(ts)partition by toYYYYMMDD(stt)order by (stt, edt);

import lombok.AllArgsConstructor;
import lombok.Data;@Data
@AllArgsConstructor
public class TradePaymentWindowBean {// 窗口起始时间String stt;// 窗口终止时间String edt;// 支付成功独立用户数Long paymentSucUniqueUserCount;// 支付成功新用户数Long paymentSucNewUserCount;// 时间戳Long ts;
}

5.2.2、创建时间工具类

为了去重，我们需要对每一条数据都设置一个时间，因为对于重复数据，它们在原始表中的时间字段值都是一样的。

FlinkSQL 提供了几个可以获取当前时间戳的函数

localtimestamp()：返回本地时区的当前时间戳，返回类型为 TIMESTAMP(3)。在流处理模式下会对每条记录计算一次时间。而在批处理模式下，仅在查询开始时计算一次时间，所有数据使用相同的时间。
current_timestamp()：返回本地时区的当前时间戳，返回类型为 TIMESTAMP_LTZ(3)。在流处理模式下会对每条记录计算一次时间。而在批处理模式下，仅在查询开始时计算一次时间，所有数据使用相同的时间。
now()：与 current_timestamp 相同。
current_row_timestamp()：返回本地时区的当前时间戳，返回类型为 TIMESTAMP_LTZ(3)。无论在流处理模式还是批处理模式下，都会对每行数据计算一次时间。

这里，我们使用current_row_timestamp 来作为时间，我们需要给订单预处理表中添加：

current_row_timestamp() as row_op_ts-- 在建表语句中添加
row_op_ts TIMESTAMP_LTZ(3)

那么，下单事务事实表来源于订单预处理表，支付成功事务事实表依赖于下单事务事实表，搜易当然也应该添加该字段。

import java.util.Comparator;public class TimestampLtz3CompareUtil {public static int compare(String timestamp1, String timestamp2) {// 数据格式 2022-04-01 10:20:47.302Z// 1. 去除末尾的时区标志，'Z' 表示 0 时区String cleanedTime1 = timestamp1.substring(0, timestamp1.length() - 1);String cleanedTime2 = timestamp2.substring(0, timestamp2.length() - 1);// 2. 提取小于 1秒的部分String[] timeArr1 = cleanedTime1.split("\\.");String[] timeArr2 = cleanedTime2.split("\\.");String microseconds1 = new StringBuilder(timeArr1[timeArr1.length - 1]).append("000").toString().substring(0, 3);String microseconds2 = new StringBuilder(timeArr2[timeArr2.length - 1]).append("000").toString().substring(0, 3);int micro1 = Integer.parseInt(microseconds1);int micro2 = Integer.parseInt(microseconds2);// 3. 提取 yyyy-MM-dd HH:mm:ss 的部分String date1 = timeArr1[0];String date2 = timeArr2[0];Long ts1 = DateFormatUtil.toTs(date1, true);Long ts2 = DateFormatUtil.toTs(date2, true);// 4. 获得精确到毫秒的时间戳long microTs1 = ts1 * 1000 + micro1;long microTs2 = ts2 * 1000 + micro2;long divTs = microTs1 - microTs2;return divTs < 0 ? -1 : divTs == 0 ? 0 : 1;}public static void main(String[] args) {System.out.println(compare("2022-04-01 11:10:55.040Z","2022-04-01 11:10:55.04Z"));}
}

5.2.3、读取DWD支付成功事务事实表

读取DWD支付成功事务事实表并转为 JSON 格式，然后按照订单明细id进行分组（为了对回撤流的数据进行去重，根据相同明细id的时间进行判断）

// TODO 3. 读取 dwd_trade_pay_detail_suc 的数据String groupId = "dws_trade_payment_suc_window";DataStreamSource<String> paymentSucDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_trade_pay_detail_suc", groupId));// TODO 4. 将数据转为JSON格式SingleOutputStreamOperator<JSONObject> jsonDS = paymentSucDS.flatMap(new RichFlatMapFunction<String, JSONObject>() {@Overridepublic void flatMap(String value, Collector<JSONObject> out) throws Exception {try {JSONObject jsonObject = JSONObject.parseObject(value);out.collect(jsonObject);} catch (Exception e) {// 可以选择输出到侧输出流e.printStackTrace();}}});// TODO 5. 按照订单明细id分组KeyedStream<JSONObject, String> keyedStream = jsonDS.keyBy(json -> json.getString("order_detail_id"));

5.2.4、状态编程对回撤流中的数据去重

这里的回撤流是因为支付成功事务事实表需要用订单明细 innner join 订单表 left join 订单明细活动 left join 订单明细活动造成的；

这里的定时器我们设置为 5 s，因为上游订单预处理表在生成的时候需要 join，我们当时设置的就是 5s，所以理论上 5s 之后，订单明细活动表和订单明细优惠券表的 left join 也应该完成了，这个状态就没必要保留了。

// TODO 6. 使用状态编程过滤最新数据输出(需要使用状态和定时器所以使用 process)SingleOutputStreamOperator<JSONObject> filterDS = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {private ValueState<JSONObject> lastPaySucDateState;@Overridepublic void open(Configuration parameters) throws Exception {lastPaySucDateState = getRuntimeContext().getState(new ValueStateDescriptor<>("last-pay-suc", JSONObject.class));}@Overridepublic void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {JSONObject state = lastPaySucDateState.value();if (state == null) {lastPaySucDateState.update(value);// 注册定时器ctx.timerService().registerEventTimeTimer(ctx.timerService().currentProcessingTime() + 5000L);} else {String stateRt = state.getString("row_op_ts");String curRt = value.getString("row_op_ts");int compare = TimestampLtz3CompareUtil.compare(stateRt, curRt);if (compare != 1) { // 状态里的时间小lastPaySucDateState.update(value);}}}@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {super.onTimer(timestamp, ctx, out);// 输出并清空状态数据JSONObject value = lastPaySucDateState.value();out.collect(value);lastPaySucDateState.clear();}});

5.2.5、提取事件时间并生成水位线

这里选择 callback_time ，它是支付成功后的回调时间；

// TODO 7. 提取事件时间生成水位线SingleOutputStreamOperator<JSONObject> jsonWithWmDS = filterDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {@Overridepublic long extractTimestamp(JSONObject element, long recordTimestamp) {return DateFormatUtil.toTs(element.getString("callback_time"), true);}}));

5.2.6、按照 user_id 分组并提取支付成功独立用户数和首次支付成功用户数

// TODO 8. 按照 user_id 分组KeyedStream<JSONObject, String> keyedByUidDS = jsonWithWmDS.keyBy(json -> json.getString("user_id"));// TODO 9. 提取独立支付成功用户数和首次支付成功用户数SingleOutputStreamOperator<TradePaymentWindowBean> tradePaymentDS = keyedByUidDS.flatMap(new RichFlatMapFunction<JSONObject, TradePaymentWindowBean>() {private ValueState<String> lastDtState;@Overridepublic void open(Configuration parameters) throws Exception {lastDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastDt", String.class));}@Overridepublic void flatMap(JSONObject value, Collector<TradePaymentWindowBean> out) throws Exception {String lastDt = lastDtState.value();String curDt = value.getString("callback_time").split(" ")[0];// 当日支付人数long pay = 0L;// 首次支付人数long newPay = 0L;// 判断状态是否为nullif (lastDt == null) {pay = 1;newPay = 1;lastDtState.update(curDt);} else if (!lastDt.equals(curDt)) {pay = 1;lastDtState.update(curDt);}// 写出if (pay == 1) {out.collect(new TradePaymentWindowBean("", "", newPay, pay, DateFormatUtil.toTs(curDt)));}}});

5.2.7、开窗聚合并写出到 clickhouse

开窗是为了实时刷新到报表，聚合依然是那两个函数：增量聚合（聚合结果），全量聚合（补充窗口起止字段）；

// TODO 10. 开窗，聚合SingleOutputStreamOperator<TradePaymentWindowBean> resultDS = tradePaymentDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10))).reduce(new ReduceFunction<TradePaymentWindowBean>() {@Overridepublic TradePaymentWindowBean reduce(TradePaymentWindowBean value1, TradePaymentWindowBean value2) throws Exception {value1.setPaymentSucNewUserCount(value1.getPaymentSucNewUserCount() + value2.getPaymentSucNewUserCount());value1.setPaymentSucUniqueUserCount(value1.getPaymentSucUniqueUserCount() + value2.getPaymentSucUniqueUserCount());return value1;}}, new AllWindowFunction<TradePaymentWindowBean, TradePaymentWindowBean, TimeWindow>() {@Overridepublic void apply(TimeWindow window, Iterable<TradePaymentWindowBean> values, Collector<TradePaymentWindowBean> out) throws Exception {TradePaymentWindowBean next = values.iterator().next();next.setTs(System.currentTimeMillis());next.setStt(DateFormatUtil.toYmdHms(window.getStart()));next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));out.collect(next);}});// TODO 11. 写出到 clickhouseresultDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_trade_payment_suc_window values(?,?,?,?,?)"));// TODO 12. 启动任务env.execute("DwsTradePaymentSucWindow");