Flink实时电商数仓（八）

用户域登录各窗口汇总表

主要任务：从kafka页面日志主题读取数据，统计
- 七日回流用户：之前活跃的用户，有一段时间不活跃了，之后又开始活跃，称为回流用户
- 当日独立用户数：同一个用户当天重复登录，只算作一个独立用户。

思路分析

读取kafka页面主题数据
转换数据结构：String -> JSONObject
过滤数据，uid不为null
- 登录的两种情况
  - 用户打开应用后自动登录
  - 用户打印应用后没有登录，浏览后跳转到登录页面
- 过滤条件：
  - uid不为null且last_page_id is null
  - last_page_id = login
设置水位线
按照uid分组
统计回流用户数和独立用户数
开窗聚合
写入doris

具体实现

设置端口、并行度、消费者组、kafka主题
读取dwd页面主题数据
- stream.print()
对数据进行清洗过滤：uid不为空
- stream.flatMap()使用flatMap过滤
- new FlatMapFunction<>(){}在该方法内部转换为JSONObject, 并且获取uid和lastPageId, try-catch这段代码
- 判断是否满足思路分析中的条件，如果中途发生异常，直接catch后打印到控制台清理掉即可。
先注册水位线
- jsonObjStream.assignTimestampAndWatermark
- new SerializableTimestampAssigner<>, 提取数据中的ts
按照uid分组
- stream.keyby()按照uid进行分组
判断独立用户和回流用户
- 创建UserLoginBean, 使用状态保存用户的登录信息
- 在open方法中，getRuntimeContext().getState(new ValueStateDescriptor<>("last_login_dt",String.class))创建状态记录用户上一次的登录时间
- 在processElement方法中比较当前登录的日期和状态存储的日期
  - 如果lastLoginDt==null是新用户
  - 如果不为空，判断上次登录时间和当前时间的差值是否大于7天；如果大于7天，说明是回流用户。
  - 如果小于7天，还需要判断上次登录时间是否是今天，如果不是今天，则说明该用户本次是独立用户。
开窗聚合
- 使用滚动窗口开窗聚合
- 在reduce算子中写聚合逻辑
- 在process算子中获取窗口信息
写入doris
- 创建doris sink，写出到doris

核心代码

public static void main(String[] args) {new DwsUserUserLoginWindow().start(10024,4,"dws_user_user_login_window", Constant.TOPIC_DWD_TRAFFIC_PAGE);}@Overridepublic void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {//1.读取dwd页面数据//stream.print();//2. 对数据进行清洗过滤SingleOutputStreamOperator<JSONObject> jsonObjStream = etl(stream);//3. 注册水位线SingleOutputStreamOperator<JSONObject> withWatermarkStream = addWatermark(jsonObjStream);//4. 按照uid分组KeyedStream<JSONObject, String> keyedStream = getKeyedStream(withWatermarkStream);//5. 判断独立用户和回流用户SingleOutputStreamOperator<UserLoginBean> processedStream = getUserLoginBeanStream(keyedStream);//processedStream.print();//开窗聚合SingleOutputStreamOperator<UserLoginBean> reducedStream = getReducedStream(processedStream);//reducedStream.print();//写入DorisreducedStream.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DWS_USER_USER_LOGIN_WINDOW));}