DWD

层剩余的需求上一节我们把日志数据根据不同类型分流写入到了不同的主题

1、流量域独立访客事务事实表

既然是独立访客就必须对日志中的数据做去重独立访客数一般用来做日活指标因为我们的机器一般都是

小时全年无休的所以我们实时数仓也可以做这种日级别的指标需求通过状态来存储历史就可以实现而怎么判断访客是否重复这就又用到了

Flink

中的状态编程状态就是历史和上一节我们判断新老访客一样我们这里也可以给每个

mid

keyby存储上一次访问的日期注意是日期只精确到天每来一条数据就判断它的

lastVisitDate

独立访客数据对应的页面必然是会话起始页面last_page_id

null所以对于跨天的访问不能计算在内昨天到今天访问了多个页面而今天页面的

last_page_id

获取执行环境StreamExecutionEnvironment

env

StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);

1.1

开启checkpointenv.enableCheckpointing(5

60000L,

CheckpointingMode.EXACTLY_ONCE);env.getCheckpointConfig().setCheckpointStorage(hdfs://hadoop102:8020/s/ck);env.getCheckpointConfig().setCheckpointTimeout(10

60000L);env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);

设置最大共存的checkpoint数量env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));

固定频率重启:

uvDetail;DataStreamSourceString

pageDS

env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic,

groupId));//

两步都一起完成SingleOutputStreamOperatorJSONObject

jsonDS

JSONObject.parseObject(value);//

last_page_idString

jsonObject.getJSONObject(page).getString(last_page_id);if

(last_page_id

json.getJSONObject(common).getString(mid));//

TODO

使用富函数,因为富函数提供更多的信息如上下文等SingleOutputStreamOperatorJSONObject

uvDS

ValueStateDescriptor(lastVisit,

getRuntimeContext().getState(stateDescriptor);}Overridepublic

boolean

更新状态lastVisitDate.update(curDate);return

true;}return

dwd_traffic_unique_visitor_detail;uvDS.map(data

data.toJSONString()).addSink(MyKafkaUtil.getFlinkKafkaProducer(targetTopic));//

TODO

执行任务env.execute(DwdTrafficUniqueVisitorDetail);}}

1.3、TTL

上面我们的代码逻辑看起来已经没什么问题了但是我们可以设想假设一个用户,2024-01-01

lastVisitDate

天我们依然要一直存储它的状态而我们判断用户是否已经登录的逻辑是lastVisitDate

lastVisitDate今天所以我们完全可以在一天之后把该用户的

lastVisitDate

ValueStateDescriptor(lastVisit,

String.class);//

TTLstateDescriptor.enableTimeToLive(new

StateTtlConfig.Builder(Time.days(1))//

TTL

可更新,并且在创建或更新状态的时候更新.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());lastVisitDate

getRuntimeContext().getState(stateDescriptor);}

2、流量域用户跳出事务事实表

跳出的概念跳出指的是用户在一次会话中只访问了一个页面的情况注意粒度是会话我们在之前做离线数仓的时候做过跳出率的指标对于离线数仓我们可以在

DWS

层构建一张流量域近1日会话粒度页面浏览表dws_traffic_session_page_view_1d通过下面的

SQL

CAST(SUM(IF(page_count1,1,0))/COUNT(*))

DECIMAL(16,2)

dws_traffic_session_page_view_1d

在这里的实时数仓中我们不可能等到一天结束最后才去计算跳出率但是我们这里又没有

思路1会话窗口

如果我短时间10s内发生多个跳出但是正好这些跳出都在一个会话这会导致窗口结束时误以为这不是跳出毕竟窗口内有多条数据②

可能我的一次正常的会话被会话窗口切分到两个不同的会话窗口结果把一个非跳出访问计算为

个跳出访问

会结束所以我们可以设置一个定时器定时器时间范围内的数据如果没数据来就视作一个会话结束触发计算并结合状态编程把新会话的首页存入状态

last_page

null则该页面是新的会话起始页开启定时器并将数据自身写入状态如果状态不为

null说明刚跳出一次并且在定时器时间范围内又进来一次这种情况需要将第一条数据跳出的数据也就是写入状态中的数据输出然后将自身写入状态定时器依然存在等时间到了触发计算如果

last_page

期望所有匹配的事件严格的一个接一个出现中间没有任何不匹配的事件。

对应方法为

next()松散连续:

忽略匹配的事件之间的不匹配的事件。

对应方法为followedBy()不确定的松散连续:

更进一步的松散连续允许忽略掉一些匹配事件的附加匹配。

对应方法为followedByAny()。

这里需要注意因为我们后面要保证数据有序所以我们最好指定事件时间的提取字段并添加水位线设置合理的超时时间理论上可以保证数据绝对有序

public

获取执行环境StreamExecutionEnvironment

env

StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);

1.1

开启checkpointenv.enableCheckpointing(5

60000L,

CheckpointingMode.EXACTLY_ONCE);env.getCheckpointConfig().setCheckpointStorage(hdfs://hadoop102:8020/s/ck);env.getCheckpointConfig().setCheckpointTimeout(10

60000L);env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);

设置最大共存的checkpoint数量env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));

固定频率重启:

user_jump_detail;DataStreamSourceString

pageDS

env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic,

groupId));//

JSONSingleOutputStreamOperatorJSONObject

jsonDS

pageDS.map(JSON::parseObject);//

TODO

jsonDS.assignTimestampsAndWatermarks(WatermarkStrategy.JSONObjectforBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new

SerializableTimestampAssignerJSONObject()

{Overridepublic

element.getLong(ts);}})).keyBy(json

json.getJSONObject(common).getString(mid));接下来是核心的定义

CEP

Pattern.JSONObjectbegin(start).where(new

{Overridepublic

value.getJSONObject(page).getString(last_page_id)

{Overridepublic

value.getJSONObject(page).getString(last_page_id)

null;}}).within(Time.seconds(10L));//

等价于

Pattern.JSONObjectbegin(start).where(new

{Overridepublic

value.getJSONObject(page).getString(last_page_id)

null;}}).times(2)

next.within(Time.seconds(10L));//

TODO

建模式序列作用到流上PatternStreamJSONObject

patternStream

OutputTag(timeout);SingleOutputStreamOperatorString

selectDS

patternStream.select(timeoutTag,//

超时数据new

PatternTimeoutFunctionJSONObject,

String()

对于超时数据来说,当前的数据第一个规则匹配上了,第二个没有匹配上导致超时,那么我们要提取的就是当前数据(第一个数据,第二个数据没来)//

Map

数据类型,因为考虑到我们可能使用的是循环模式(只有一个key)Overridepublic

String

map.get(start).get(0).toJSONString();}},

PatternSelectFunctionJSONObject,

String()

匹配上的数据,我们只要第一个数据,因为只能证明第一个数据是跳出数据Overridepublic

String

map.get(start).get(0).toJSONString();}});DataStreamString

timeoutDS

selectDS.getSideOutput(timeoutTag);//

TODO

dwd_traffic_user_jump_detail;unionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(targetTopic));//

TODO

启动任务env.execute(DwdTrafficUserJumpDetail);}

last_page_id

超时时间内规则一被满足未等到第二条数据则会被判定为超时数据。

所以我们只要把超时数据和

last_page_id