• 数仓开发之DWD层（一）

    目录

     一：流量域未经加工的事务事实表

    1.1 主要任务

    1.2 思路

    1.3 图解

     1.4 代码

    二：流量域独立访客事务事实表

    2.1 主要任务

    2.2 思路分析

     2.3 图解

     2.4 代码

    DWD层设计要点：

    （1）DWD层的设计依据是维度建模理论，该层存储维度模型的事实表。

    （2）DWD层表名的命名规范为dwd_数据域_表名

     一：流量域未经加工的事务事实表

    1.1 主要任务

    1）数据清洗（ETL）

    数据传输过程中可能会出现部分数据丢失的情况，导致 JSON 数据结构不再完整，因此需要对脏数据进行过滤。

    2）新老访客状态标记修复

    日志数据 common 字段下的 is_new 字段是用来标记新老访客状态的，1 表示新访客，0 表示老访客。前端埋点采集到的数据可靠性无法保证，可能会出现老访客被标记为新访客的问题，因此需要对该标记进行修复。

    3）分流

    本节将通过分流对日志数据进行拆分，生成五张事务事实表写入 Kafka

      • 流量域页面浏览事务事实表
      • 流量域启动事务事实表
      • 流量域动作事务事实表
      • 流量域曝光事务事实表
      • 流量域错误事务事实表

    1.2 思路

    1）数据清洗（ETL）

    对流中数据进行解析，将字符串转换为 JSONObject，如果解析报错则必然为脏数据。

    定义侧输出流，将脏数据发送到侧输出流，写入 Kafka 脏数据主题。

    2）新老访客状态标记修复

    （1）前端埋点新老访客状态标记设置规则

    以神策提供的第三方埋点服务中新老访客状态标记设置规则为例

    • Web 端：用户第一次访问埋入神策 SDK 页面的当天（即第一天），JS SDK 会在网页的 cookie 中设置一个首日访问的标记，并设置第一天 24 点之前，该标记为 true，即第一天触发的网页端所有事件中，is_new = 1。第一天之后，该标记则为 false，即第一天之后触发的网页端所有事件中，is_new = 0；
    • 小程序端：用户第一天访问埋入神策 SDK 的页面时，小程序 SDK 会在 storage 缓存中创建一个首日为 true 的标记，并且设置第一天 24 点之前，该标记均为 true。即第一天触发的小程序端所有事件中，is_new = 1。第一天之后，该标记则为 false，即第一天之后触发的小程序端所有事件中，​is_new = 0；
    • APP 端：用户安装 App 后，第一次打开埋入神策 SDK 的 App 的当天，Android/iOS SDK 会在手机本地缓存内，创建一个首日为 true 的标记，并且设置第一天 24 点之前，该标记均为 true。即第一天触发的 APP 端所有事件中，is_new = 1。第一天之后，该标记则为 false，即第一天之后触发的 APP 端所有事件中，is_new = 0。

    本项目模拟生成的是 APP 端日志数据。对于此类日志，如果首日之后用户清除了手机本地缓存中的标记，再次启动 APP 会重新设置一个首日为 true 的标记，导致本应为 0 的 is_new 字段被置为 1，可能会给相关指标带来误差。因此，有必要对新老访客状态标记进行修复。

    （2）新老访客状态标记修复思路

    运用 Flink 状态编程，为每个 mid 维护一个键控状态，记录首次访问日期。

    ①如果 is_new 的值为 1

    a）如果键控状态为 null，认为本次是该访客首次访问 APP，将日志中 ts 对应的日期更新到状态中，不对 is_new 字段做修改；

    b）如果键控状态不为 null，且首次访问日期不是当日，说明访问的是老访客，将 is_new 字段置为 0；

    c）如果键控状态不为 null，且首次访问日期是当日，说明访问的是新访客，不做操作；

    ②如果 is_new 的值为 0

    a）如果键控状态为 null，说明访问 APP 的是老访客但本次是该访客的页面日志首次进入程序。当前端新老访客状态标记丢失时，日志进入程序被判定为老访客，Flink 程序就可以纠正被误判的访客状态标记，只要将状态中的日期设置为今天之前即可。本程序选择将状态更新为昨日；

    b）如果键控状态不为 null，说明程序已经维护了首次访问日期，不做操作。

    3）利用侧输出流实现数据拆分

    （1）埋点日志结构分析

    前端埋点获取的 JSON 字符串（日志）可能存在 common、start、page、displays、actions、err、ts 七种字段。其中

    • common 对应的是公共信息，是所有日志都有的字段
    • err 对应的是错误信息，所有日志都可能有的字段
    • start 对应的是启动信息，启动日志才有的字段
    • page 对应的是页面信息，页面日志才有的字段
    • displays 对应的是曝光信息，曝光日志才有的字段，曝光日志可以归为页面日志，因此必然有 page 字段
    • actions 对应的是动作信息，动作日志才有的字段，同样属于页面日志，必然有 page 字段。动作信息和曝光信息可以同时存在。
    • ts 对应的是时间戳，单位：毫秒，所有日志都有的字段

    综上，我们可以将前端埋点获取的日志分为两大类：启动日志和页面日志。二者都有 common 字段和 ts 字段，都可能有 err 字段。页面日志一定有 page 字段，一定没有 start 字段，可能有 displays 和 actions 字段；启动日志一定有 start 字段，一定没有 page、displays 和 actions 字段。

    （2）分流日志分类

    本节将按照内容，将日志分为以下五类

    • 启动日志
    • 页面日志
    • 曝光日志
    • 动作日志
    • 错误日志

    （3）分流思路

    ①所有日志数据都可能拥有 err 字段，所有首先获取 err 字段，如果返回值不为 null 则将整条日志数据发送到错误侧输出流。然后删掉 JSONObject 中的 err 字段及对应值；

    ②判断是否有 start 字段，如果有则说明数据为启动日志，将其发送到启动侧输出流；如果没有则说明为页面日志，进行下一步；

    ③页面日志必然有 page 字段、 common 字段和 ts 字段，获取它们的值，ts 封装为包装类 Long，其余两个字段的值封装为 JSONObject；

    ④判断是否有 displays 字段，如果有，将其值封装为 JSONArray，遍历该数组，依次获取每个元素（记为 display），封装为JSONObject。创建一个空的 JSONObject，将 display、common、page和 ts 添加到该对象中，获得处理好的曝光数据，发送到曝光侧输出流。动作日志的处理与曝光日志相同（注意：一条页面日志可能既有曝光数据又有动作数据，二者没有任何关系，因此曝光数据不为 null 时仍要对动作数据进行处理）；

    ⑤动作日志和曝光日志处理结束后删除 displays 和 actions 字段，此时主流的 JSONObject 中只有 common 字段、 page 字段和 ts 字段，即为最终的页面日志。

    处理结束后，页面日志数据位于主流，其余四种日志分别位于对应的侧输出流，将五条流的数据写入 Kafka 对应主题即可。

    1.3 图解

     1.4 代码

    1）在 KafkaUtil 工具类中补充 getKafkaProducer() 方法 --- 生产者

    1.  public static FlinkKafkaProducer getFlinkKafkaProducer(String topic){
    2.         return new FlinkKafkaProducer(KAFKA_SERVER,
    3.                 topic,
    4.                 new SimpleStringSchema());
    5.     }

    2）创建 DateFormatUtil 工具类用于日期格式化

    1. package com.atguigu.gmall.realtime.util;
    2.  
    3. import java.time.LocalDateTime;
    4. import java.time.LocalTime;
    5. import java.time.ZoneId;
    6. import java.time.ZoneOffset;
    7. import java.time.format.DateTimeFormatter;
    8. import java.util.Date;
    9.  
    10. public class DateFormatUtil {
    11.  
    12.     private static final DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd");
    13.     private static final DateTimeFormatter dtfFull = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    14.  
    15.     public static Long toTs(String dtStr, boolean isFull) {
    16.         
    17.         LocalDateTime localDateTime = null;
    18.         if (!isFull) {
    19.             dtStr = dtStr + " 00:00:00";
    20.         }
    21.         localDateTime = LocalDateTime.parse(dtStr, dtfFull);
    22.  
    23.         return localDateTime.toInstant(ZoneOffset.of("+8")).toEpochMilli();
    24.     }
    25.  
    26.     public static Long toTs(String dtStr) {
    27.         return toTs(dtStr, false);
    28.     }
    29.  
    30.     public static String toDate(Long ts) {
    31.         Date dt = new Date(ts);
    32.         LocalDateTime localDateTime = LocalDateTime.ofInstant(dt.toInstant(), ZoneId.systemDefault());
    33.         return dtf.format(localDateTime);
    34.     }
    35.  
    36.     public static String toYmdHms(Long ts) {
    37.         Date dt = new Date(ts);
    38.         LocalDateTime localDateTime = LocalDateTime.ofInstant(dt.toInstant(), ZoneId.systemDefault());
    39.         return dtfFull.format(localDateTime);
    40.     }
    41.  
    42.     public static void main(String[] args) {
    43.         System.out.println(toYmdHms(System.currentTimeMillis()));
    44.     }
    45. }

     3）主程序

    1. package com.atguigu.app.dwd;
    2.  
    3. import com.alibaba.fastjson.JSON;
    4. import com.alibaba.fastjson.JSONArray;
    5. import com.alibaba.fastjson.JSONObject;
    6. import com.atguigu.utils.DateFormatUtil;
    7. import com.atguigu.utils.MyKafkaUtil;
    8. import org.apache.flink.api.common.functions.RichMapFunction;
    9. import org.apache.flink.api.common.state.ValueState;
    10. import org.apache.flink.api.common.state.ValueStateDescriptor;
    11. import org.apache.flink.configuration.Configuration;
    12. import org.apache.flink.streaming.api.datastream.DataStream;
    13. import org.apache.flink.streaming.api.datastream.DataStreamSource;
    14. import org.apache.flink.streaming.api.datastream.KeyedStream;
    15. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    16. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    17. import org.apache.flink.streaming.api.functions.ProcessFunction;
    18. import org.apache.flink.util.Collector;
    19. import org.apache.flink.util.OutputTag;
    20.  
    21. import java.util.concurrent.TimeUnit;
    22.  
    23. public class BaseLogApp {
    24.  
    25.     //数据源：web/app -> Nginx -> 日志服务器（.log） -> flume ->Kafka (ODS) -> FlinkApp -> Kafka(DWD)
    26.     //程  序：Mock(lg.sh) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK)
    27.  
    28.     public static void main(String[] args) throws Exception {
    29.         //1.获取执行环境
    30.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    31.         env.setParallelism(1);
    32.         /*
    33.         
    34.         //1.1 开启CheckPoint
    35.         env.enableCheckpointing(5 *6000L , CheckpointingMode.EXACTLY_ONCE);
    36.         env.getCheckpointConfig().setCheckpointTimeout(10 *6000L);
    37.         env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
    38.         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));
    39.         
    40.         //1.2 设置状态后端
    41.         env.setStateBackend(new HashMapStateBackend());
    42.         env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop107:8020/211126/ck");
    43.         System.setProperty("HADOOP_USER_NAME","atguigu");
    44.         
    45.          */
    46.  
    47.         //2.消费Kafka topic_log 主题的数据创建流
    48.  
    49.         String topic = "topic_log";
    50.         String groupId = "base_log_app_211126";
    51.         DataStreamSource kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
    52.  
    53.         //3.过滤掉非JSON格式的数据&将每行数据转换为JSON对象
    54.  
    55.         OutputTag dirtyTag = new OutputTag("dirty"){
    56.         };
    57.         SingleOutputStreamOperator jsonObjDS = kafkaDS.process(new ProcessFunction() {
    58.             @Override
    59.             public void processElement(String value, ProcessFunction.Context context, Collector collector) throws Exception {
    60.                 try {
    61.                     JSONObject jsonObject = JSON.parseObject(value);
    62.                     collector.collect(jsonObject);//主输出流
    63.                 } catch (Exception e) {
    64.                     context.output(dirtyTag, value);//非JSON数据输出到侧输出流中
    65.                 }
    66.             }
    67.         });
    68.         //获取侧输出流脏数据并打印
    69.         DataStream dirtyDS = jsonObjDS.getSideOutput(dirtyTag);
    70.         dirtyDS.print("Dirty>>>>>>>");
    71.  
    72.         //将脏数据输出到指定主题上
    73.         String dirty_topic="dwd_traffic_dirty_log";
    74.         dirtyDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(dirty_topic));
    75.  
    76.         //4.按照Mid分组
    77.         KeyedStream keyedStream = jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"));
    78.  
    79.         //5.使用状态编程做新老用户标记校验
    80.         SingleOutputStreamOperator jsonObjectWithNewFlagDS = keyedStream.map(new RichMapFunction() {
    81.             private ValueState lastVisitState;
    82.  
    83.             @Override
    84.             public void open(Configuration parameters) throws Exception {
    85.                 lastVisitState = getRuntimeContext().getState(new ValueStateDescriptor("last-visit", String.class));
    86.             }
    87.  
    88.             @Override
    89.             public JSONObject map(JSONObject value) throws Exception {
    90.                 //获取is_new 标记 & ts 并将时间戳转换为年月日
    91.                 String isNew = value.getJSONObject("common").getString("is_new");
    92.                 Long ts = value.getLong("ts");
    93.                 String curDate = DateFormatUtil.toDate(ts);
    94.  
    95.                 //获取状态中的日期
    96.                 String lastDate = lastVisitState.value();
    97.  
    98.                 //判断is_new标记是否为“1”
    99.                 if ("1".equals(isNew)) {
    100.  
    101.                     if (lastDate == null) {
    102.                         lastVisitState.update(curDate);
    103.                     } else if (!lastDate.equals(curDate)) {
    104.                         value.getJSONObject("common").put("is_new", "0");
    105.                     }
    106.  
    107.                 } else if (lastDate == null) {
    108.                     lastVisitState.update(DateFormatUtil.toDate(ts - 24 * 60 * 60 * 1000L));//将时间改成昨天
    109.                 }
    110.  
    111.                 return value;
    112.             }
    113.         });
    114.  
    115.         //6.使用测输出流进行分流处理,这里把页面当成主流，其他（启动、曝光、动作、错误）放到侧输出流中
    116.         OutputTag startTag = new OutputTag("start"){
    117.         };
    118.         OutputTag displayTag = new OutputTag("display"){
    119.         };
    120.         OutputTag actionTag = new OutputTag("action"){
    121.         };
    122.         OutputTag errorTag = new OutputTag("error"){
    123.         };
    124.         SingleOutputStreamOperator pageDS = jsonObjectWithNewFlagDS.process(new ProcessFunction() {
    125.             @Override
    126.             public void processElement(JSONObject value, ProcessFunction.Context context, Collector collector) throws Exception {
    127.                 //尝试获取错误信息
    128.                 String err = value.getString("err");
    129.                 if (err != null) {
    130.                     //将数据写到侧输出流
    131.                     context.output(errorTag, value.toJSONString());
    132.                 }
    133.  
    134.                 //移除错误信息
    135.                 value.remove("err");
    136.  
    137.                 //尝试获得启动信息
    138.                 String start = value.getString("start");
    139.                 if (start != null) {
    140.                     //将数据写到start侧输出流
    141.                     context.output(startTag, value.toJSONString());
    142.                 } else {
    143.  
    144.                     //获取公共（common）信息 & 页面 & 时间戳（ts）
    145.                     String common = value.getString("common");
    146.                     String pageId = value.getJSONObject("page").getString("page_id");
    147.                     Long ts = value.getLong("ts");
    148.  
    149.                     //尝试获取动作数据
    150.                     JSONArray actions = value.getJSONArray("actions");
    151.                     if (actions != null && actions.size() > 0) {//避免actions标签内容为空
    152.  
    153.                         //遍历曝光数据 & 写到display侧输出流中
    154.                         for (int i = 0; i < actions.size(); i++) {
    155.                             JSONObject action = actions.getJSONObject(i);
    156.                             action.put("common", common);
    157.                             action.put("page_id", pageId);
    158.                             context.output(actionTag, action.toJSONString());
    159.                         }
    160.  
    161.                     }
    162.  
    163.                     //尝试获取曝光数据
    164.                     JSONArray displays = value.getJSONArray("displays");
    165.                     if (displays != null && displays.size() > 0) {//避免displays标签内容为空
    166.  
    167.                         //遍历曝光数据 & 写到display侧输出流中
    168.                         for (int i = 0; i < displays.size(); i++) {
    169.                             JSONObject display = displays.getJSONObject(i);
    170.                             display.put("common", common);
    171.                             display.put("page_id", pageId);
    172.                             display.put("ts", ts);
    173.                             context.output(displayTag, display.toJSONString());
    174.                         }
    175.  
    176.                     }
    177.  
    178.                     //移除曝光和动作数据 & 写到页面日志主流
    179.                     value.remove("displays");
    180.                     value.remove("actions");
    181.                     collector.collect(value.toJSONString());
    182.                 }
    183.             }
    184.         });
    185.  
    186.         //7.提取各个侧输出流数据
    187.         DataStream startDS = pageDS.getSideOutput(startTag);
    188.         DataStream displayDS = pageDS.getSideOutput(displayTag);
    189.         DataStream actionDS = pageDS.getSideOutput(actionTag);
    190.         DataStream errorDS = pageDS.getSideOutput(errorTag);
    191.  
    192.         //8.将数据打印写入对应的主题
    193.         pageDS.print("Page>>>>>>>>>");
    194.         startDS.print("Start>>>>>>>");
    195.         displayDS.print("Display>>>");
    196.         actionDS.print("Action>>>>>");
    197.         errorDS.print("Error>>>>>>>");
    198.  
    199.         String page_topic = "dwd_traffic_page_log";
    200.         String start_topic = "dwd_traffic_start_log";
    201.         String display_topic = "dwd_traffic_display_log";
    202.         String action_topic = "dwd_traffic_action_log";
    203.         String error_topic = "dwd_traffic_error_log";
    204.  
    205.         pageDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
    206.         startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
    207.         displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
    208.         actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
    209.         errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));
    210.  
    211.         //9.启动任务
    212.         env.execute("BaseLogApp");
    213.     }
    214. }

    二：流量域独立访客事务事实表

    2.1 主要任务

    过滤页面数据中的独立访客访问记录。

    2.2 思路分析

    1）过滤 last_page_id 不为 null 的数据

    独立访客数据对应的页面必然是会话起始页面，last_page_id 必为 null。过滤 last_page_id != null 的数据，减小数据量，提升计算效率。

    2）筛选独立访客记录

    运用 Flink 状态编程，为每个 mid 维护一个键控状态，记录末次登录日期。

    如果末次登录日期为 null 或者不是今日，则本次访问是该 mid 当日首次访问，保留数据，将末次登录日期更新为当日。否则不是当日首次访问，丢弃数据。

    3）状态存活时间设置

    如果保留状态，第二日同一 mid 再次访问时会被判定为新访客，如果清空状态，判定结果相同，所以只要时钟进入第二日状态就可以清空。

    设置状态的 TTL 为 1 天，更新模式为 OnCreateAndWrite，表示在创建和更新状态时重置状态存活时间。如：2022-02-21 08:00:00 首次访问，若 2022-02-22 没有访问记录，则 2022-02-22 08:00:00 之后状态清空。

     2.3 图解

     2.4 代码

    1）主程序

    1. package com.atguigu.app.dwd;
    2.  
    3. import com.alibaba.fastjson.JSON;
    4. import com.alibaba.fastjson.JSONAware;
    5. import com.alibaba.fastjson.JSONObject;
    6. import com.atguigu.utils.DateFormatUtil;
    7. import com.atguigu.utils.MyKafkaUtil;
    8. import org.apache.flink.api.common.functions.FlatMapFunction;
    9. import org.apache.flink.api.common.functions.RichFilterFunction;
    10. import org.apache.flink.api.common.state.StateTtlConfig;
    11. import org.apache.flink.api.common.state.ValueState;
    12. import org.apache.flink.api.common.state.ValueStateDescriptor;
    13. import org.apache.flink.api.common.time.Time;
    14. import org.apache.flink.configuration.Configuration;
    15. import org.apache.flink.streaming.api.datastream.DataStreamSource;
    16. import org.apache.flink.streaming.api.datastream.KeyedStream;
    17. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    18. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    19. import org.apache.flink.util.Collector;
    20.  
    21. //数据源：web/app -> Nginx -> 日志服务器（.log） -> flume ->Kafka (ODS) -> FlinkApp -> Kafka(DWD) -> Flink(App) -> Kafka(DWD)
    22. //程  序：Mock(lg.sh) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK) -> DwdTrafficUniqueVisitorDetail ->Kafka(ZK)
    23.  
    24. public class DwdTrafficUniqueVisitorDetail {
    25.     //独立访客需求（uv）
    26.     public static void main(String[] args) throws Exception {
    27.         //1.创建执行环境
    28.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    29.         env.setParallelism(1);
    30.  
    31.         /*
    33.         //1.1 开启CheckPoint
    34.         env.enableCheckpointing(5 *6000L , CheckpointingMode.EXACTLY_ONCE);
    35.         env.getCheckpointConfig().setCheckpointTimeout(10 *6000L);
    36.         env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
    37.         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));
    39.         //1.2 设置状态后端
    40.         env.setStateBackend(new HashMapStateBackend());
    41.         env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop107:8020/211126/ck");
    42.         System.setProperty("HADOOP_USER_NAME","atguigu");
    44.          */
    45.  
    46.         //2.读取kafka页面日志主题创建流
    47.  
    48.         String topic= "dwd_traffic_page_log";
    49.        // String topic= "topic_log";
    50.         String groupId= "unique_visitor_detail_211126";
    51.         DataStreamSource kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
    52.  
    53.         //3.过滤掉上一跳页面不为null的数据并将每行数据转换为JSON对象
    54.         SingleOutputStreamOperator jsonObjDS = kafkaDS.flatMap(new FlatMapFunction() {
    55.             @Override
    56.             public void flatMap(String value, Collector collector) throws Exception {
    57.                 try {
    58.                     JSONObject jsonObject = JSON.parseObject(value);
    59.                     //获取上一跳页面ID
    60.                     String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
    61.                     if (lastPageId == null) {//收集当天首次登陆的数据
    62.                         collector.collect(jsonObject);
    63.                     }
    64.                 } catch (Exception e) {
    65.                     e.printStackTrace();
    66.                     System.out.println("脏数据>>>"+value);//脏数据
    67.                 }
    68.  
    69.             }
    70.         });
    71.  
    72.         //4.按照mid分组
    73.         KeyedStream keyedStream = jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"));
    74.  
    75.         //5.使用状态编程实现按照mid的去重
    76.         SingleOutputStreamOperator uvDS = keyedStream.filter(new RichFilterFunction() {
    77.  
    78.             private ValueState lastVisitState;
    79.  
    80.             @Override
    81.             public void open(Configuration parameters) throws Exception {
    82.                 ValueStateDescriptor stateDescriptor = new ValueStateDescriptor<>("last-visit", String.class);
    83.  
    84.                 //设置状态的TTL
    85.                 StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1))
    86.                         .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    87.                         .build();
    88.                 stateDescriptor.enableTimeToLive(ttlConfig);
    89.  
    90.                 lastVisitState = getRuntimeContext().getState(stateDescriptor);
    91.             }
    92.  
    93.             @Override
    94.             public boolean filter(JSONObject value) throws Exception {
    95.  
    96.                 //获取状态数据&当前数据中的时间戳并转换为日期
    97.                 String lastDate = lastVisitState.value();
    98.                 Long ts = value.getLong("ts");
    99.                 String curDate = DateFormatUtil.toDate(ts);
    100.                 //如果最近一次访问app/web的日期不是今天 或者 没有最近一次访问的日期（新用户） ，则将最近一次的访问日期
    101.                 // 更新到今天，并存储到流中
    102.                 if (lastDate == null || !lastDate.equals(curDate)) {
    103.                     lastVisitState.update(curDate);
    104.                     return true;
    105.                 }
    106.  
    107.                 return false;
    108.             }
    109.         });
    110.  
    111.         //6.将数据写入kafka
    112.         String targetTopic = "dwd_traffic_unique_visitor_detail";
    113.         uvDS.print(">>>>>>>>>>");
    114.         uvDS.map(JSONAware::toJSONString)
    115.                 .addSink(MyKafkaUtil.getFlinkKafkaProducer(targetTopic));
    116.  
    117.         //7.执行
    118.         env.execute();
    119.     }
    120. }

  • 相关阅读:
    58同城2024届校招后端研发一面面经
    UVM 的精髓在于给验证人员提供了快速搭建 testbench 的途径
    Java SE 17 新增特性
    Electron常见问题 64 - Electron的升级安装包会下载到本地哪个目录？
    监督学习的介绍
    从源码分析vue3组件的生命周期
    剖析虚幻渲染体系（15）- XR专题
    初识设计模式 - 职责链模式
    【Nuget】程序包源
    ＜Linux进程概念＞——《Linux》
  • 原文地址：https://blog.csdn.net/JiaXingNashishua/article/details/127811540