4、实时数仓DWM层业务实现

约 16245 字大约 54 分钟

2026-01-17

DWM 层的设计

设计思路

我们在之前通过分流等手段，把数据分拆成了独立的 Kafka Topic。那么接下来如何处理数据，就要思考一下我们到底要通过实时计算出哪些指标项。

因为实时计算与离线不同，实时计算的开发和运维成本都是非常高的，要结合实际情况考虑是否有必要象离线数仓一样，建一个大而全的中间层。

如果没有必要大而全，这时候就需要大体规划一下要实时计算出的指标需求了。把这些指标以主题宽表的形式输出就是我们的 DWS 层。

需求梳理

为什么在实时数仓中没有DWT层，因为DWT层存放的是历史的聚集的结果，实时数仓中不需要这一层，可以实时计算。

当然实际需求还会有更多，这里主要以为可视化大屏为目的进行实时计算的处理。

DWM 层的定位是什么，DWM 层主要服务 DWS，因为部分需求直接从 DWD 层到DWS 层中间会有一定的计算量，而且这部分计算的结果很有可能被多个 DWS 层主题复用，

所以部分 DWD 成会形成一层 DWM，我们这里主要涉及业务。

访问 UV 计算
跳出明细计算
订单宽表
支付宽表

业务属于

用户
用户以设备为判断标准，在移动统计中，每个独立设备认为是一个独立用户。Android系统根据IMEI号，IOS系统根据OpenUDID来标识一个独立用户，每部手机一个用户。
新增用户
首次联网使用应用的用户。如果一个用户首次打开某APP，那这个用户定义为新增用户；卸载再安装的设备，不会被算作一次新增。新增用户包括日新增用户、周新增用户、月新增用户。
活跃用户
打开应用的用户即为活跃用户，不考虑用户的使用情况。每天一台设备打开多次会被计为一个活跃用户。
周（月）活跃用户
某个自然周（月）内启动过应用的用户，该周（月）内的多次启动只记一个活跃用户。
月活跃率
月活跃用户与截止到该月累计的用户总和之间的比例。
沉默用户
用户仅在安装当天（次日）启动一次，后续时间无再启动行为。该指标可以反映新增用户质量和用户与APP的匹配程度。
版本分布
不同版本的周内各天新增用户数，活跃用户数和启动次数。利于判断APP各个版本之间的优劣和用户行为习惯。
本周回流用户
上周未启动过应用，本周启动了应用的用户。
连续n周活跃用户
连续n周，每周至少启动一次。
忠实用户
连续活跃5周以上的用户
连续活跃用户
连续2周及以上活跃的用户
近期流失用户
连续n(2<= n <= 4)周没有启动应用的用户。（第n+1周没有启动过）
留存用户
某段时间内的新增用户，经过一段时间后，仍然使用应用的被认作是留存用户；这部分用户占当时新增用户的比例即是留存率。
例如，5月份新增用户200，这200人在6月份启动过应用的有100人，7月份启动过应用的有80人，8月份启动过应用的有50人；则5月份新增用户一个月后的留存率是50%，二个月后的留存率是40%，三个月后的留存率是25%。
用户新鲜度
每天启动应用的新老用户比例，即新增用户数占活跃用户数的比例。
单次使用时长
每次启动使用的时间长度。
日使用时长
累计一天内的使用时间长度。
启动次数计算标准
IOS平台应用退到后台就算一次独立的启动；Android平台我们规定，两次启动之间的间隔小于30秒，被计算一次启动。用户在使用过程中，若因收发短信或接电话等退出应用30秒又再次返回应用中，那这两次行为应该是延续而非独立的，所以可以被算作一次使用行为，即一次启动。业内大多使用30秒这个标准，但用户还是可以自定义此时间间隔。

DWM层访客 UV 计算

需求分析与思路

UV，全称是 Unique Visitor，即独立访客，对于实时计算中，也可以称为 DAU(Daily Active User)，即每日活跃用户，因为实时计算中的 UV 通常是指当日的访客数。

那么如何从用户行为日志中识别出当日的访客，那么有两点：

其一，是识别出该访客打开的第一个页面，表示这个访客开始进入我们的应用

那具体在这里我们根据哪一个字段进行判断呢？

我们根据日志中的last_page_id判断，如果该字段是null，那么说明是今天第一次登录该页面，否则不空，也就是说不是第一次。这个字段代表上一跳的地址，为null说明没有上一跳。

其二，由于访客可以在一天中多次进入应用，所以我们要在一天的范围内进行去重，因为用户每一次访问页面，都会产生一个访问日志。

因为我们要统计日活跃量，而每一个访客可以多次重复登录，所以需要进行去重操作。我们可以使用Flink中的keyStated，一个mid对应于一个状态。key1-->state,状态中可以存储年月日时间。

数据流：模拟生成数据->日志处理服务器->写到 kafka 的 ODS 层（ods_base_log） ->BaseLogApp分流->dwd_page_log->UniqueVisitApp 读取输出

代码实现

从 Kafka 的 dwd_page_log 主题接收数据

 //TODO 2.读取Kafka dwd_page_log 主题的数据
        String groupId = "unique_visit_app";//消费者组
        String sourceTopic = "dwd_page_log";//dwd层数据源，是kafka中的一个主题
        String sinkTopic = "dwm_unique_visit";//存放在kafka中的dwm_unique_visit主题中
//        读取数据
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtils.getKafkaConsumer(sourceTopic, groupId));

        //TODO 3.将每行数据转换为JSON对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);

核心的过滤代码实现

首先用 keyby 按照 mid 进行分组，每组表示当前设备的访问情况
分组后使用 keystate 状态，记录用户进入时间，实现 RichFilterFunction 完成过滤
重写 open 方法用来初始化状态
重写 filter 方法进行过滤
- 可以直接筛掉 last_page_id 不为空的字段，因为只要有上一页，说明这条不是这个用户进入的首个页面。
- 状态用来记录用户的进入时间，只要这个 lastVisitDate 是今天，就说明用户今天已经访问过了所以筛除掉。如果为空或者不是今天，说明今天还没访问过，则保留。
- 因为状态值主要用于筛选是否今天来过，所以这个记录过了今天基本上没有用了，这里 enableTimeToLive 设定了 1 天的过期时间，避免状态过大。

       //TODO 4.过滤数据  状态编程  只保留每个mid每天第一次登陆的数据
//        首先进行过滤分组，这里过滤掉的是mid为null的数据，也就是不合法的数据
        KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

//        因为这里需要用到状态编程，所以使用富函数，普通的Filter不可以使用状态
        /**
         * 在这里我们选用什么状态呢》valueState即可，因为我们存储的是一个时间
         */
        SingleOutputStreamOperator<JSONObject> uvDS = keyedStream.filter(new RichFilterFunction<JSONObject>() {

//            时间存储围殴String了欸行
            private ValueState<String> dateState;
//            因为数据里面只有时间戳，所以我们需要进行转换
            private SimpleDateFormat simpleDateFormat;

            /**
             * 初始化
             * @param parameters
             * @throws Exception
             */
            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("date-state", String.class);


                /**
                 * Flink中的状态可以设置一个超时时间
                 */
                //设置状态的超时时间以及更新时间的方式
                StateTtlConfig stateTtlConfig = new StateTtlConfig
                        .Builder(Time.hours(24))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build();

//                设置状态过时时间
                valueStateDescriptor.enableTimeToLive(stateTtlConfig);

//              访问时间状态
                dateState = getRuntimeContext().getState(valueStateDescriptor);

                simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
            }

            @Override
            public boolean filter(JSONObject value) throws Exception {

                //取出上一条页面信息
                String lastPageId = value.getJSONObject("page").getString("last_page_id");

                //判断上一条页面是否为Null
                if (lastPageId == null || lastPageId.length() <= 0) {

                    //取出状态数据
                    String lastDate = dateState.value();

                    //取出今天的日期，也就是数据中的ts字段
                    String curDate = simpleDateFormat.format(value.getLong("ts"));

                    //判断两个日期是否相同
                    if (!curDate.equals(lastDate)) {
//                        如果不相同，那么就保留当前数据，更新状态
//                        注意，这里的状态更新，是一条数据对应一个状态，在这之前已经按照Mid进行分组了
                        dateState.update(curDate);
                        return true;
                    }
//                    else {
//                        return false;
//                    }
                }
//                上一条不是null，直接返回false，过滤掉即可
                return false;

            }
        });

将过滤处理后的 UV写入到 Kafka的 dwm_unique_visit

        //TODO 5.将数据写入Kafka
        uvDS.print();
        uvDS.map(JSONAware::toJSONString)//将json转换为String类型
                .addSink(MyKafkaUtils.getKafkaProducer(sinkTopic));

DWM 层跳出明细计算

需求分析与思路

什么是跳出

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

使用会话窗口的方案解决，在没有会话id的时候，如何确定这个数据是同一次会话中访问的呢？什么时候使用会话窗口，会话窗口之间的间隔时间我们自己可以确定，在我我们需要使用会话id去计算某一些指标的时候，但是这个时候没有会话id，那么我们就可以使用会话窗口解决。如果两次会话之间相隔时间较长，那么就认为是一次新的会话。

会话窗口会产生的问题

但是会话窗口可能导致丢失数据，比如数据A和数据B之间的间隔不足10秒，并且不是同一条数据，那么针对开的10秒会话窗口，两条数据分到一个窗口中，会发生丢失（因为我们是根据两条数据之间的时间间隔；来判断他们是否属于同一个窗口，现在两条数据属于统一个窗口，我们认为他没有跳出，但是实际上他是跳出的），针对这种情况，我们使用CEP。

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

计算跳出行为的思路

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

该页面是用户近期访问的第一个页面，这个可以通过该页面是否有上一个页面（last_page_id）来判断，如果这个表示为空，就说明这是这个访客这次访问的第一个页面。

首次访问之后很长一段时间（自己设定），用户没继续再有其他页面的访问。，如果有会话id的话，我们可以使用会话id方便的解决。

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

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

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

如果有会话id，那么就取出会话id，并且这条会话数据只有一条就可以了，但是没有会话id怎么办？
可以使用会话窗口，当没有会话id的时候，如何确定某一条数据是同一次会话中访问的呢？
会话窗口应用场景就是我们想根据会话id计算某一些指标的时候，但是没有会话id，我们就可以使用会话窗口代替。根据时间，如果两次会话时间间隔太长，那么就认为是两次会话。我们可以规定一个间隔时间。最后对会话窗口计算的时候，我们可以使用全量窗口进行计算，就是使用apply()函数，如果是一次访问，那么会话窗口中就只有一条数据。
但是这种思路会丢失数据，比如会话窗口间隔10秒，当两条数据a和b都是一次单跳，进入页面然后出去，那么当两条数据进入同一个窗口的时候，数据不止一条，就不认为是单跳。
使用cep，进来一条数据，只要判断其上一条是空，下一跳是还是null，说明这个数据就是我们需要的。但是这里也要定义一个超时时间，比如当前事件吓一跳事件迟迟不来怎么版，所以我们需要定义超时事件，将超时事件放到测输出流。

CEP编程三步骤

定义模式序列
将模式序列应用到流上
提取匹配上的超时事件

cep可以处理乱序数据

代码实现

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

//TODO 2.读取Kafka主题的数据创建流
        String sourceTopic = "dwd_page_log";
        String groupId = "userJumpDetailApp";
        String sinkTopic = "dwm_user_jump_detail";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtils.getKafkaConsumer(sourceTopic, groupId));

        //TODO 3.将每行数据转换为JSON对象并提取时间戳生成Watermark
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject)
                .assignTimestampsAndWatermarks(WatermarkStrategy
//                        设置乱序时间
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
//                            提取事件时间
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("ts");
                            }
                        }));

通过 Flink 的 CEP 完成跳出判断

确认添加了 CEP 的依赖包
设定时间语义为事件时间并指定数据中的 ts 字段为事件时间

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

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

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

env.setStreamTimeCharacteristic(TimeCharacteristic. EventTime)

根据日志数据的 mid 进行分组

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

  //TODO 5.将模式序列作用到流上
        PatternStream<JSONObject> patternStream = CEP
                .pattern(jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"))//需要针对同一个mid左同一件事，所以使用keyby
                        , pattern);

配置 CEP 表达式

 //TODO 4.定义模式序列
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                String lastPageId = value.getJSONObject("page").getString("last_page_id");
//                我们需要保留lastPageId为null的数据，也就是第一次访问页面的用户
                return lastPageId == null || lastPageId.length() <= 0;
            }
        }).next("next").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                String lastPageId = value.getJSONObject("page").getString("last_page_id");
                return lastPageId == null || lastPageId.length() <= 0;
            }
//            设置会话窗口时间间隔
        }).within(Time.seconds(10));
 //使用循环模式  定义模式序列
        Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                String lastPageId = value.getJSONObject("page").getString("last_page_id");
                return lastPageId == null || lastPageId.length() <= 0;
            }
        })
                .times(2)
                .consecutive() //指定严格近邻(next)
                .within(Time.seconds(10));

根据表达式筛选流

 //TODO 5.将模式序列作用到流上
        PatternStream<JSONObject> patternStream = CEP
                .pattern(jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"))//需要针对同一个mid左同一件事，所以使用keyby
                        , pattern);

提取命中的数据

设定超时时间标识 timeoutTag。
flatSelect 方法中，实现 PatternFlatTimeoutFunction 中的 timeout 方法。
所有 out.collect 的数据都被打上了超时标记。
本身的 flatSelect 方法提取匹配上的数据。
通过 SideOutput 侧输出流输出超时数据

//TODO 6.提取匹配上的和超时事件
        OutputTag<JSONObject> timeOutTag = new OutputTag<JSONObject>("timeOut") {
        };

        SingleOutputStreamOperator<JSONObject> selectDS = patternStream.select(timeOutTag,
                new PatternTimeoutFunction<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject timeout(Map<String, List<JSONObject>> map, long ts) throws Exception {
                        return map.get("start").get(0);
                    }
                }, new PatternSelectFunction<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
                        return map.get("start").get(0);
                    }
                });

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

  提取超时时间的数据，超时时间的数据放在测输出流中
        DataStream<JSONObject> timeOutDS = selectDS.getSideOutput(timeOutTag);

        //TODO 7.UNION两种事件
        DataStream<JSONObject> unionDS = selectDS.union(timeOutDS);

        //TODO 8.将数据写入Kafka
        unionDS.print();
        unionDS.map(JSONAware::toJSONString)
                .addSink(MyKafkaUtils.getKafkaProducer(sinkTopic));

Flink join

Flink一共又4中join。

Interval不需要开窗就可以join。

带开窗函数的join操作

join:连接的其他数据流
whrere:第一个数据流的键
equalTo:第二个数据流的key.
window:开窗
apply:join函数，也就是连接函数。

带窗口的join操作

Tumbling Window Join

如果Flink使用的是滚动窗口，那么就和spark streaming是一样的。先收集完窗口中的数据，然后做统一计算。也就是说赞一个批次，处理完后向下传输。可以说也是一种微批次处理。

这种方式下没有重复的数据，因为窗口不会发生重叠。

DataStream<Integer> orangeStream = ...
DataStream<Integer> greenStream = ...

orangeStream.join(greenStream)
    .where(<KeySelector>)
    .equalTo(<KeySelector>)
    .window(TumblingEventTimeWindows.of(Time.milliseconds(2)))
    .apply (new JoinFunction<Integer, Integer, String> (){
        @Override
        public String join(Integer first, Integer second) {
            return first + "," + second;
        }
    });

Sliding Window Join

在滑动窗口中，可能会输出重复数据，因为窗口又重叠的部分，所以有一部分数据在多个窗口中进行关联输出。，比如图中的0，0，他们属于两个窗口，所以在每一个窗口中都会输出一次。

DataStream<Integer> orangeStream = ...
DataStream<Integer> greenStream = ...

orangeStream.join(greenStream)
    .where(<KeySelector>)
    .equalTo(<KeySelector>)
    .window(SlidingEventTimeWindows.of(Time.milliseconds(2) /* size */, Time.milliseconds(1) /* slide */))
    .apply (new JoinFunction<Integer, Integer, String> (){
        @Override
        public String join(Integer first, Integer second) {
            return first + "," + second;
        }
    });

Session Window Join

滚动窗口和滑动窗口的时间都是对齐的，对于会话窗口，时间是不对其的，也就是要求两个流相隔gap时间都没有数据流来才可以，两个流都同时满足超时时间才可以，中间之后某一个流满足时间间隔是不可以的。而滚动和滑动窗口的两个流都是同时开始，同时结束。

DataStream<Integer> orangeStream = ...
DataStream<Integer> greenStream = ...

orangeStream.join(greenStream)
    .where(<KeySelector>)
    .equalTo(<KeySelector>)
    .window(EventTimeSessionWindows.withGap(Time.milliseconds(1)))
    .apply (new JoinFunction<Integer, Integer, String> (){
        @Override
        public String join(Integer first, Integer second) {
            return first + "," + second;
        }
    });

Interval Join

不需要开窗

在一条流中的某一个时间点去join另外一条流中的某一个时间区间，如果有满足join条件的数据，全部输出。

比如上图中所示，橙色数据流在2位置开始join，此时绿色数据流相对橙色2位置左侧的数据保存在状态中，但是绿色数据流相对橙色数据流2右侧数据还没有到，那么此时会把橙色数据流2进行状态保存，等到绿色数据流右侧数据全部到齐，做join操作。

关联另一条数据流中一个范围内的数据时候，有一个时间上下界，为什么可以关联到某个时间点之前的数据呢，也就是时间下界的数据？

这是因为使用了状态编程，会把某个时间点之前到时间下界之间的数据写入状态中，

那为什么可以join某一个时间点到时间上界之间的数据呢？

也是状态编程，会把某一个时间点到时间上界之间的数据保存一段时间，然后join操作。

在Flink中有状态，所以可以保存历史数据，也就是状态，但是在spark streamming中没有状态，所以写入的是redis数据库保存，然后在关联。

The interval join currently only supports event time.

DataStream<Integer> orangeStream = ...
DataStream<Integer> greenStream = ...

orangeStream
    .keyBy(<KeySelector>)
    .intervalJoin(greenStream.keyBy(<KeySelector>))
    .between(Time.milliseconds(-2), Time.milliseconds(1))
    .process (new ProcessJoinFunction<Integer, Integer, String(){

        @Override
        public void processElement(Integer left, Integer right, Context ctx, Collector<String> out) {
            out.collect(first + "," + second);
        }
    });

keyBy:指出橙色主流按照哪一个字段进行join。
intervalJoin：绿色流按照哪一个字段join。
between：join的时间上下界。
process：处理方法

DWM 层-支付宽表

需求分析和思路

再支付表中，支付行为主要和订单相关，那么我们希望根据商品来计算其总金额，也就是被支付的数量，被支付的次数，但是支付表中是和订单相关，并没有商品的明细，所以我们需要做一个支付宽表。

支付宽表的目的，最主要的原因是支付表没有到订单明细，支付金额没有细分到商品上，没有办法统计商品级的支付状况。

所以本次宽表的核心就是要把支付表的信息与订单宽表关联上

解决方案有两个

一个是把订单宽表输出到 HBase上，在支付宽表计算时查询 HBase，这相当于把订单宽表作为一种维度进行管理。
一个是用流的方式接收订单宽表，然后用双流 join 方式进行合并。因为订单与支付产生有一定的时差。所以必须用 intervalJoin 来管理流的状态时间，保证当支付到达时订单宽表还保存在状态中。

在这里我们选用第二种方式，因为订单和支付中间不是连续的，可能下订单了，但是过了15分钟后支付，而这需要保存为一个状态，如果这个状态使用双流join，只需要将状态保存15分钟即可，但是Hbase默认是永久保存的，但是这里并不需要永久保存状态。

另一个原因是如果把订单宽表作为维度表去查询，因为这个表很大，查询延迟必然很高。

使用kafka双流Join效率高，实现起来相对容易。订单宽表本来是在kafka中，没必要再写入hbase中作为宽表处理，这样增加了难度和复杂度。

创建支付实体类 PaymentInfo

@Data
public class PaymentInfo {
    Long id;
    Long order_id;
    Long user_id;
    BigDecimal total_amount;
    String subject;
    String payment_type;
    String create_time;
    String callback_time;
}

创建支付宽表实体类 PaymentWide

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.beanutils.BeanUtils;

import java.lang.reflect.InvocationTargetException;
import java.math.BigDecimal;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class PaymentWide {

    Long payment_id;
    String subject;
    String payment_type;
    String payment_create_time;
    String callback_time;
    Long detail_id;
    Long order_id;
    Long sku_id;
    BigDecimal order_price;
    Long sku_num;
    String sku_name;
    Long province_id;
    String order_status;
    Long user_id;
    BigDecimal total_amount;
    BigDecimal activity_reduce_amount;
    BigDecimal coupon_reduce_amount;
    BigDecimal original_total_amount;
    BigDecimal feight_fee;
    BigDecimal split_feight_fee;
    BigDecimal split_activity_amount;
    BigDecimal split_coupon_amount;
    BigDecimal split_total_amount;
    String order_create_time;

    String province_name;   //查询维表得到
    String province_area_code;
    String province_iso_code;
    String province_3166_2_code;

    Integer user_age;       //用户信息
    String user_gender;

    Long spu_id;           //作为维度数据 要关联进来
    Long tm_id;
    Long category3_id;
    String spu_name;
    String tm_name;
    String category3_name;

    public PaymentWide(PaymentInfo paymentInfo, OrderWide orderWide) {
        mergeOrderWide(orderWide);
        mergePaymentInfo(paymentInfo);
    }

    public void mergePaymentInfo(PaymentInfo paymentInfo) {
        if (paymentInfo != null) {
            try {
                BeanUtils.copyProperties(this, paymentInfo);
                payment_create_time = paymentInfo.create_time;
                payment_id = paymentInfo.id;
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            } catch (InvocationTargetException e) {
                e.printStackTrace();
            }
        }
    }

    public void mergeOrderWide(OrderWide orderWide) {
        if (orderWide != null) {
            try {
                BeanUtils.copyProperties(this, orderWide);
                order_create_time = orderWide.create_time;
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            } catch (InvocationTargetException e) {
                e.printStackTrace();
            }
        }
    }
}

支付宽表处理程序

//数据流：web/app -> nginx -> SpringBoot -> Mysql -> FlinkApp -> Kafka(ods) -> FlinkApp -> Kafka/Phoenix(dwd-dim) -> FlinkApp(redis) -> Kafka(dwm) -> FlinkApp -> Kafka(dwm)
//程  序：         MockDb               -> Mysql -> FlinkCDC -> Kafka(ZK) -> BaseDbApp -> Kafka/Phoenix(zk/hdfs/hbase) -> OrderWideApp(Redis) -> Kafka -> PaymentWideApp -> Kafka
public class PaymentWideApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //1.1 设置CK&状态后端
        //env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall-flink-210325/ck"));
        //env.enableCheckpointing(5000L);
        //env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //env.getCheckpointConfig().setCheckpointTimeout(10000L);
        //env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);

        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart());

        //TODO 2.读取Kafka主题的数据创建流 并转换为JavaBean对象 提取时间戳生成WaterMark
        String groupId = "payment_wide_group";
        String paymentInfoSourceTopic = "dwd_payment_info";
        String orderWideSourceTopic = "dwm_order_wide";
        String paymentWideSinkTopic = "dwm_payment_wide";

//        订单数据流
        SingleOutputStreamOperator<OrderWide> orderWideDS = env.addSource(MyKafkaUtils.getKafkaConsumer(orderWideSourceTopic, groupId))
                .map(line -> JSON.parseObject(line, OrderWide.class))
//                forMonotonousTimestam表示时间戳是增长的方式
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderWide>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderWide>() {
                            @Override
                            public long extractTimestamp(OrderWide element, long recordTimestamp) {
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                try {
                                    return sdf.parse(element.getCreate_time()).getTime();
                                } catch (ParseException e) {
                                    e.printStackTrace();
                                    return recordTimestamp;
                                }
                            }
                        }));
//        支付数据流
        SingleOutputStreamOperator<PaymentInfo> paymentInfoDS = env.addSource(MyKafkaUtils.getKafkaConsumer(paymentInfoSourceTopic, groupId))
                .map(line -> JSON.parseObject(line, PaymentInfo.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<PaymentInfo>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<PaymentInfo>() {
                            @Override
                            public long extractTimestamp(PaymentInfo element, long recordTimestamp) {
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                try {
                                    return sdf.parse(element.getCreate_time()).getTime();
                                } catch (ParseException e) {
                                    e.printStackTrace();
                                    return recordTimestamp;
                                }
                            }
                        }));

        //TODO 3.双流JOIN
        SingleOutputStreamOperator<PaymentWide> paymentWideDS = paymentInfoDS.keyBy(PaymentInfo::getOrder_id)
                .intervalJoin(orderWideDS.keyBy(OrderWide::getOrder_id))
                .between(Time.minutes(-15), Time.seconds(5))//给5s的时间延迟
                .process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                    @Override
                    public void processElement(PaymentInfo paymentInfo, OrderWide orderWide, Context ctx, Collector<PaymentWide> out) throws Exception {
                        out.collect(new PaymentWide(paymentInfo, orderWide));
                    }
                });

        //TODO 4.将数据写入Kafka，写入的是支付宽表主题：dwm_payment_wide
        paymentWideDS.print(">>>>>>>>>");
        paymentWideDS
                .map(JSONObject::toJSONString)
                .addSink(MyKafkaUtils.getKafkaProducer(paymentWideSinkTopic));

        //TODO 5.启动任务
        env.execute("PaymentWideApp");

    }

}

封装日期转换工具类

public class DateTimeUtil {

    private final static DateTimeFormatter formater = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    public static String toYMDhms(Date date) {
        LocalDateTime localDateTime = LocalDateTime.ofInstant(date.toInstant(), ZoneId.systemDefault());
        return formater.format(localDateTime);
    }

    public static Long toTs(String YmDHms) {
        LocalDateTime localDateTime = LocalDateTime.parse(YmDHms, formater);
        return localDateTime.toInstant(ZoneOffset.of("+8")).toEpochMilli();
    }
}

因为SimpleDateFormat不是一个线程安全的类，如果使用SimpleDateFormat类，那么需要把这个操作放在一个方法内部，让他变为一个局部变量，否则放在方法外面，就存在线程安全问题。

LocalDateTime是一个线程安全的类，全局只有一个对象，也可以保证线程安全，但是上面的SimpleDateFormat类，全局只有一个就存在线程安全问题。

小结

使用技术

学会利用状态（state）进行去重操作。（需求：UV 计算）
学会利用 CEP 可以针对一组数据进行筛选判断。需求：跳出行为计算
学会使用 intervalJoin 处理流 join
学会处理维度关联，并通过旁路缓存和异步查询对其进行性能优化。

再DWM层，我们形成了四张宽表：

访客主题宽表（离线数仓按照user_id，实时按照mid来）
跳出明细主题宽表
商品主题宽表（也就是订单宽表）
支付宽表

形成三张宽表，主要是为ads层统计服务

实现类说明

DimUtil

封装查询维度的工具类 DimUtil，在查询维度信息的时候，只有表名字和我们传输的过滤条件不同，其他的sql基本都一样，所以我们再做一次封装。

RedisUtil

实现Rides缓存类。因为直接访问hbase进行数据的查询延迟非常高，如果连接不关闭，大概处理一条数据是13毫秒，也就是单并行度，一秒钟大概处理80条数据，80次访问，所以使用Rides进行优化。

在删除数据的时候，我们先删除Rides中数据，然后写入hbase中，这样做主要保证数据一致性。

由于是两个不同的进程，如果再把Rides中的数据删除之后，那orderWideApp刚好又查询了一次，那么又把查询到的老数据写入Rides，这个时候，还没有向Hbase中写入数据。此时Rides中还是老的数据。

先删除Rides中数据，在改hbase中数据，在删除Rides中数据，这种方式一定程度上可以解决问题，如果任务挂掉的情况，Rides中数据没有删除掉，还不行。
我们Hbase中存储的是维度数据，那么维度数据一般是缓慢变化的，更新操作并不是很多，所以这块可以直接不删除Rides中数据，直接向Rides中写一份修改后的数据即可，这种方案最好。如果写入hbase失败怎么办，因为Rides中数据是保存24小时的，即使保存到hbase中的任务失败，那么我们可以重新启动任务，在24小时内重新写入hbase即可。所以解决了hbase中间出错的问题，如果Rides失败，那么直接去hbase中查询数据即可，Rides启动之后，再写入Rides即可，这样即使中间某一方出错，其他来查询也不会出现问题。
为什么不使用事务或者锁方案，因为在Rides中锁是乐观锁，又处理请求的话，会释放锁。

我们一共6个维度表，假设6个维度表数据全部再Rides中，那么一个维度的查询需要1毫秒，6个维度需要6毫秒，我们假设5毫秒，那么就是说，每一秒中，单个并行度可以处理200条数据，这样就不会产生反压。如果数据超过200每一秒，那么就会产生反压。再高峰期大概每一秒1000-2000条，所以还需要对这种方案进行优化。

目前Rides方案是满足要求，但是我们需要考虑可拓展性，比如再搞活动的时候，或者kafka做压测的时候，最小是2000，所以我们还需要优化。

我们一般不会对Rides加锁，因为Rides是乐观锁。

获取单例的线程池对象 corePoolSize:指定了线程池中的线程数量，它的数量决定了添加的任务是开辟新的线程去执行，还是放到 workQueue 任务队列中去； maximumPoolSize:指定了线程池中的最大线程数量，这个参数会根据你使用的 workQueue 任务队列的类型，决定线程池会开辟的最大线程数量； keepAliveTime:当线程池中空闲线程数量超过 corePoolSize 时，多余的线程会在多长时间内被销毁； unit:keepAliveTime 的单位 workQueue:任务队列，被添加到线程池中，但尚未被执行的任务

4、实时数仓DWM层业务实现

详情