前言
今天的任务是完成流量域最后一个需求、用户域的两个需求以及交易域的部分需求；
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>() {
@Override
public 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>() {
@Override
public 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;
@Override
public 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);
// 设置 TTL
homeStateDescriptor.enableTimeToLive(ttlConfig);
detailStateDescriptor.enableTimeToLive(ttlConfig);
homeLastVisit = getRuntimeContext().getState(homeStateDescriptor);
detailLastVisit = getRuntimeContext().getState(detailStateDescriptor);
}
@Override
public 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>() {
@Override
public 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>() {
@Override
public 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. 写入到 clickhouse
resultDS.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>() {
@Override
public 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>() {
@Override
public 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;
@Override
public 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);
}
@Override
public 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>() {
@Override
public 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. 写入到 clickhouse
resultDS.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. 将动态表转为流并写入到 clickhouse
DataStream<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>() {
@Override
public 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;
@Override
public 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);
}
@Override
public 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>() {
@Override
public CartAddUuBean reduce(CartAddUuBean value1, CartAddUuBean value2) throws Exception {
value1.setCartAddUuCt(value1.getCartAddUuCt() + value2.getCartAddUuCt());
return value1;
}
}, new AllWindowFunction<CartAddUuBean, CartAddUuBean, TimeWindow>() {
@Override
public 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. 写出到 clickhouse
resultDS.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: a1
SELECT ...
FROM
order_detail od
join
order_info oi
on
od.order_id = oi.id
left join
order_detail_activity oa
on
od.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>() {
@Override
public 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;
@Override
public void open(Configuration parameters) throws Exception {
lastPaySucDateState = getRuntimeContext().getState(new ValueStateDescriptor<>("last-pay-suc", JSONObject.class));
}
@Override
public 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);
}
}
}
@Override
public 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>() {
@Override
public 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;
@Override
public void open(Configuration parameters) throws Exception {
lastDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastDt", String.class));
}
@Override
public 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;
// 判断状态是否为null
if (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>() {
@Override
public 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>() {
@Override
public 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. 写出到 clickhouse
resultDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_trade_payment_suc_window values(?,?,?,?,?)"));
// TODO 12. 启动任务
env.execute("DwsTradePaymentSucWindow");
总结
今天的 DWS 层到此为止，剩下了还有几个需求估计还得 1~2 天完成，这一块要比之前都难一些，争取这周日前把实时数仓完结；然后下周开始把离线和实时再好好复习一遍；
本文转载自: https://blog.csdn.net/m0_64261982/article/details/140945921
版权归原作者 让线程再跑一会 所有， 如有侵权，请联系我们删除。
版权归原作者 让线程再跑一会 所有， 如有侵权，请联系我们删除。