一 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，查看输出结果。