IT数码 购物 网址 头条 软件 日历 阅读 图书馆
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓
图片批量下载器
↓批量下载图片,美女图库↓
图片自动播放器
↓图片自动播放器↓
一键清除垃圾
↓轻轻一点,清除系统垃圾↓
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁
 
   -> 大数据 -> 【实时数仓】DWM层跳出明细计算之需求分析、读取数据、通过Flink的CEP完成跳出判断、写回kafka、测试 -> 正文阅读

[大数据]【实时数仓】DWM层跳出明细计算之需求分析、读取数据、通过Flink的CEP完成跳出判断、写回kafka、测试

一 DWM层-跳出明细计算

1 需求分析与思路

(1)什么是跳出

跳出: 用户成功访问了网站的一个页面后就退出,不再继续访问网站的其它页面。而跳出率就是用跳出次数除以访问次数。

关注跳出率,可以看出从某几个网站引流过来的访客是否能很快的被吸引,渠道引流过来的用户之间的质量对比,对于应用优化前后跳出率的对比也能看出优化改进的成果。

(2)计算跳出行为的思路

首先要识别哪些是跳出行为,要把这些跳出的访客最后一个访问的页面识别出来。那么要抓住几个特征:

  • 该页面是用户近期访问的第一个页面(新的会话)

    这个可以通过该页面是否有上一个页面(last_page_id)来判断,如果这个表示为空,就说明这是这个访客这次访问的第一个页面。

  • 首次访问之后很长一段时间(自己设定,一般为30min),用户没继续再有其他页面的访问。

这第一个特征的识别很简单,保留last_page_id为空的就可以了。但是第二个访问的判断,其实有点麻烦,首先这不是用一条数据就能得出结论的,需要组合判断,要用一条存在的数据和不存在的数据进行组合判断。而且要通过一个不存在的数据求得一条存在的数据。更麻烦的它并不是永远不存在,而是在一定时间范围内不存在。那么如何识别有一定失效的组合行为呢?

最简单的办法就是Flink自带的复杂事件处理(CEP)技术。CEP非常适合通过多条数据组合来识别某个事件。

用户跳出事件,本质上就是一个条件事件加一个超时事件的组合。

(3)实现思路

实现思路如下:

在这里插入图片描述

  • 从kafka中读取数据。
  • 使用CEP对数据进行过滤。
    • 该页面是用户近期访问的第一个页面。
    • 如果在指定时间内,有当前设备对网站其他页面的访问,说明发生了跳转。
    • 反之,则发生了跳出。
    • 可以使用within指定匹配的时间;涉及到了时间,flink1.12默认的时间语义就是事件时间语义,需要指定watermark以及提取事件时间字段。
  • 使用CEP编程步骤
    • 定义pattern
    • 将pattern应用到流上
    • 从流中按照指定的模式提取数据

在这里插入图片描述

2 读取数据

从kafka的dwd_page_log主题中读取页面日志。

(1)代码编写

/**
 * 用户跳出明细统计
 */
public class UserJumpDetailAPP {
    public static void main(String[] args) throws Exception {
        //TODO 1 基本环境准备
        //1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2 检查点设置(略)

        //TODO 3 从kafka中读取数据
        //3.1 声明消费主题以及消费者组
        String topic = "dwd_page_log";
        String groupId = "user_jump_detail_app_group";
        //3.2 获取kafka消费者对象
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 读取数据封装流
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);

        //TODO 4 对读取的数据进行类型转换 String -> JSONObject
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);
        jsonObjDS.print(">>>");

        env.execute();
    }
}

(2)测试

启动相关进程,模拟日志生成,查看是否可以正常接收到数据。

3 通过Flink的CEP完成跳出判断

(1)确认添加了CEP的依赖包

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-cep_${scala.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

(2)设定时间语义为事件时间并指定数据中的ts字段为事件时间

由于这里涉及到时间的判断,所以必须设定数据流的EventTime和水位线。这里没有设置延迟时间,实际生产情况可以视乱序情况增加一些延迟。

增加延迟把forMonotonousTimestamps换为forBoundedOutOfOrderness即可。

注意:flink1.12默认的时间语义就是事件时间,所以不需要执行。

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

// TODO 5.指定watermark以及提取事件时间字段
SingleOutputStreamOperator<JSONObject> jsonObjWithWatermark = jsonObjDS.assignTimestampsAndWatermarks(
        WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                                return jsonObj.getLong("ts");
                            }
                        }
                )
);

(3)根据日志数据的mid进行分组

因为用户的行为都是要基于相同的Mid的行为进行判断,所以要根据Mid进行分组。

// TODO 6 按照mid进行分组
KeyedStream<JSONObject, String> keyedDS = jsonObjWithWatermark.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

(4)配置CEP表达式

跳出 or 跳转的3个条件

  • 必须是一个新会话,lastPageId为空
  • 访问了其他页面
  • 不能超过一定时间
// TODO 7 定义pattern
Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("first").where(
        // 条件1:开启一个新的会话访问
        new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObj) throws Exception {
                String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                if (lastPageId == null || lastPageId.length() == 0) {
                    return true;
                }
                return false;
            }
        }
).next("second").where(
        // 条件2:访问了网站的其他页面
        new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObj) throws Exception {
                String pageId = jsonObj.getJSONObject("page").getString("page_id");
                if (pageId != null && pageId.length() > 0) {
                    return true;
                }
                return false;
            }
        }
)
        // within方法:定文匹配模式的事件序列出现的最大时间间隔。
        // 如果未完成的事件序列超过了这个事件,就会被丢弃:
        .within(Time.seconds(10));

(5)根据表达式筛选流

// TODO 8 将pattern应用到流上
PatternStream<JSONObject> patternDS = CEP.pattern(keyedDS, pattern);

(6)提取命中的数据

a 从模式中提取

在获得到一个PatternStream之后,可以应用各种转换来发现事件序列。推荐使用PatternProcessFunction

PatternProcessFunction有一个processMatch的方法在每找到一个匹配的事件序列时都会被调用。 它按照Map<String, List<IN>>的格式接收一个匹配,映射的键是模式序列中的每个模式的名称,值是被接受的事件列表(IN是输入事件的类型)。 模式的输入事件按照时间戳进行排序。为每个模式返回一个接受的事件列表的原因是当使用循环模式(比如oneToMany()times())时, 对一个模式会有不止一个事件被接受。

官网说明:从模式中提取

b 处理超时的部分匹配

当一个模式上通过within加上窗口长度后,部分匹配的事件序列就可能因为超过窗口长度而被丢弃。可以使用TimedOutPartialMatchHandler接口 来处理超时的部分匹配。这个接口可以和其它的混合使用。也就是说你可以在自己的PatternProcessFunction里另外实现这个接口。 TimedOutPartialMatchHandler提供了另外的processTimedOutMatch方法,这个方法对每个超时的部分匹配都会调用。

c 便捷的API

前面提到的PatternProcessFunction是在Flink 1.8之后引入的,从那之后推荐使用这个接口来处理匹配到的结果。 用户仍然可以使用像select/flatSelect这样旧格式的API,它们会在内部被转换为PatternProcessFunction

使用方法:

  • 设定超时时间标识 timeoutTag。
  • flatSelect方法中,实现PatternFlatTimeoutFunction中的timeout方法。
  • 所有out.collect的数据都被打上了超时标记。
  • 本身的flatSelect方法因为不需要未超时的数据所以不接受数据。
  • 通过SideOutput侧输出流输出超时数据。
        // TODO 9 从流中提取数据
        //9.1 定义侧输出流标记,FlinkCDC会将超时数据匹配放到侧数据流中
        OutputTag<String> timeoutTag = new OutputTag<>("timeoutTag");
        // 9.2 提取数据
//        patternDS.select(
//                timeoutTag,
//                new PatternTimeoutFunction<JSONObject, String>() {
//                    @Override
//                    public String timeout(Map<String, List<JSONObject>> pattern, long timestamp) throws Exception {
//                        return null;
//                    }
//                },
//                new PatternSelectFunction<JSONObject, String>() {
//                    @Override
//                    public String select(Map<String, List<JSONObject>> map) throws Exception {
//                        return null;
//                    }
//                }
//        );

        SingleOutputStreamOperator<String> resDS = patternDS.flatSelect(
                timeoutTag,
                // 处理超时数据 -- 跳出,需要进行统计
                new PatternFlatTimeoutFunction<JSONObject, String>() {
                    @Override
                    public void timeout(Map<String, List<JSONObject>> pattern, long timestamp, Collector<String> out) throws Exception {
                        List<JSONObject> jsonObjectList = pattern.get("first");
                        for (JSONObject jsonObj : jsonObjectList) {
                            out.collect(jsonObj.toJSONString());
                        }
                    }
                },
                // 处理完全匹配的数据 -- 跳转,不在此需求统计范围之内
                new PatternFlatSelectFunction<JSONObject, String>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> map, Collector<String> collector) throws Exception {

                    }
                }
        );

        // 9.3 从侧输出流中获取超时数据(跳出)
        DataStream<String> jumpDS = resDS.getSideOutput(timeoutTag);
        
        jumpDS.print(">>>>");

(7)利用测试数据完成测试

{common:{mid:101},page:{page_id:home},ts:10000} ,
{common:{mid:102},page:{page_id:home},ts:12000},
{common:{mid:102},page:{page_id:good_list,last_page_id:home},ts:15000},
{common:{mid:102},page:{page_id:good_list,last_page_id:detail},ts:30000} 
        //3.3 读取数据封装流
//        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);
        DataStream<String> kafkaDS = env
                .fromElements(
                        "{\"common\":{\"mid\":\"101\"},\"page\":{\"page_id\":\"home\"},\"ts\":10000} ",
                        "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"home\"},\"ts\":12000}",
                        "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
                                "\"home\"},\"ts\":15000} ",
                        "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
                                "\"detail\"},\"ts\":30000} "
                );

输出结果:

>>>>:3> {"common":{"mid":"101"},"page":{"page_id":"home"},"ts":10000}

4 写回kafka

将跳出数据写回到kafka的DWM层。

// TODO 10 将跳出明细写到kafka的dwm层主题
jumpDS.addSink(MyKafkaUtil.getKafkaSink("dwm_user_jump_detail"));

5 测试

将测试数据注释掉,打开kafka数据源。

打开zookeeper、kafka、日志采集服务、kafka消费者、BaseLogApp、UserJumpDetailAPP,查看输出结果。

  大数据 最新文章
实现Kafka至少消费一次
亚马逊云科技:还在苦于ETL?Zero ETL的时代
初探MapReduce
【SpringBoot框架篇】32.基于注解+redis实现
Elasticsearch:如何减少 Elasticsearch 集
Go redis操作
Redis面试题
专题五 Redis高并发场景
基于GBase8s和Calcite的多数据源查询
Redis——底层数据结构原理
上一篇文章      下一篇文章      查看所有文章
加:2022-12-25 11:16:02  更:2022-12-25 11:20:00 
 
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁

360图书馆 购物 三丰科技 阅读网 日历 万年历 2024年2日历 -2024/2/29 6:02:45-

图片自动播放器
↓图片自动播放器↓
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓
一键清除垃圾
↓轻轻一点,清除系统垃圾↓
图片批量下载器
↓批量下载图片,美女图库↓
  网站联系: qq:121756557 email:121756557@qq.com  IT数码