数仓开发之DWD层（一）

水深无声 2023-09-23 14:58 203阅读 0赞

目录

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

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 图解

b9bcd87528f04fffa6d5fbcdfe425ac8.png

1.4 代码

1）在 KafkaUtil**工具类中补充 get**KafkaProducer() 方法 —- 生产者

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

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

  1. package com.atguigu.gmall.realtime.util;
  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;
  10. public class DateFormatUtil {
  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");
  15.     public static Long toTs(String dtStr, boolean isFull) {
  17.         LocalDateTime localDateTime = null;
  18.         if (!isFull) {
  19.             dtStr = dtStr + " 00:00:00";
  20.         }
  21.         localDateTime = LocalDateTime.parse(dtStr, dtfFull);
  23.         return localDateTime.toInstant(ZoneOffset.of("+8")).toEpochMilli();
  24.     }
  26.     public static Long toTs(String dtStr) {
  27.         return toTs(dtStr, false);
  28.     }
  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.     }
  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.     }
  42.     public static void main(String[] args) {
  43.         System.out.println(toYmdHms(System.currentTimeMillis()));
  44.     }
  45. }

3**）主程序**

  1. package com.atguigu.app.dwd;
  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;
  21. import java.util.concurrent.TimeUnit;
  23. public class BaseLogApp {
  25.     //数据源：web/app -> Nginx -> 日志服务器（.log） -> flume ->Kafka (ODS) -> FlinkApp -> Kafka(DWD)
  26.     //程  序：Mock(lg.sh) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK)
  28.     public static void main(String[] args) throws Exception {
  29.         //1.获取执行环境
  30.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  31.         env.setParallelism(1);
  32.         /*
  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));
  40.         //1.2 设置状态后端
  41.         env.setStateBackend(new HashMapStateBackend());
  42.         env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop107:8020/211126/ck");
  43.         System.setProperty("HADOOP_USER_NAME","atguigu");
  45.          */
  47.         //2.消费Kafka topic_log 主题的数据创建流
  49.         String topic = "topic_log";
  50.         String groupId = "base_log_app_211126";
  51.         DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
  53.         //3.过滤掉非JSON格式的数据&将每行数据转换为JSON对象
  55.         OutputTag<String> dirtyTag = new OutputTag<String>("dirty"){
  56.         };
  57.         SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
  58.             @Override
  59.             public void processElement(String value, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> 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<String> dirtyDS = jsonObjDS.getSideOutput(dirtyTag);
  70.         dirtyDS.print("Dirty>>>>>>>");
  72.         //将脏数据输出到指定主题上
  73.         String dirty_topic="dwd_traffic_dirty_log";
  74.         dirtyDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(dirty_topic));
  76.         //4.按照Mid分组
  77.         KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"));
  79.         //5.使用状态编程做新老用户标记校验
  80.         SingleOutputStreamOperator<JSONObject> jsonObjectWithNewFlagDS = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {
  81.             private ValueState<String> lastVisitState;
  83.             @Override
  84.             public void open(Configuration parameters) throws Exception {
  85.                 lastVisitState = getRuntimeContext().getState(new ValueStateDescriptor<String>("last-visit", String.class));
  86.             }
  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);
  95.                 //获取状态中的日期
  96.                 String lastDate = lastVisitState.value();
  98.                 //判断is_new标记是否为“1”
  99.                 if ("1".equals(isNew)) {
  101.                     if (lastDate == null) {
  102.                         lastVisitState.update(curDate);
  103.                     } else if (!lastDate.equals(curDate)) {
  104.                         value.getJSONObject("common").put("is_new", "0");
  105.                     }
  107.                 } else if (lastDate == null) {
  108.                     lastVisitState.update(DateFormatUtil.toDate(ts - 24 * 60 * 60 * 1000L));//将时间改成昨天
  109.                 }
  111.                 return value;
  112.             }
  113.         });
  115.         //6.使用测输出流进行分流处理,这里把页面当成主流，其他（启动、曝光、动作、错误）放到侧输出流中
  116.         OutputTag<String> startTag = new OutputTag<String>("start"){
  117.         };
  118.         OutputTag<String> displayTag = new OutputTag<String>("display"){
  119.         };
  120.         OutputTag<String> actionTag = new OutputTag<String>("action"){
  121.         };
  122.         OutputTag<String> errorTag = new OutputTag<String>("error"){
  123.         };
  124.         SingleOutputStreamOperator<String> pageDS = jsonObjectWithNewFlagDS.process(new ProcessFunction<JSONObject, String>() {
  125.             @Override
  126.             public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
  127.                 //尝试获取错误信息
  128.                 String err = value.getString("err");
  129.                 if (err != null) {
  130.                     //将数据写到侧输出流
  131.                     context.output(errorTag, value.toJSONString());
  132.                 }
  134.                 //移除错误信息
  135.                 value.remove("err");
  137.                 //尝试获得启动信息
  138.                 String start = value.getString("start");
  139.                 if (start != null) {
  140.                     //将数据写到start侧输出流
  141.                     context.output(startTag, value.toJSONString());
  142.                 } else {
  144.                     //获取公共（common）信息 & 页面 & 时间戳（ts）
  145.                     String common = value.getString("common");
  146.                     String pageId = value.getJSONObject("page").getString("page_id");
  147.                     Long ts = value.getLong("ts");
  149.                     //尝试获取动作数据
  150.                     JSONArray actions = value.getJSONArray("actions");
  151.                     if (actions != null && actions.size() > 0) {//避免actions标签内容为空
  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.                         }
  161.                     }
  163.                     //尝试获取曝光数据
  164.                     JSONArray displays = value.getJSONArray("displays");
  165.                     if (displays != null && displays.size() > 0) {//避免displays标签内容为空
  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.                         }
  176.                     }
  178.                     //移除曝光和动作数据 & 写到页面日志主流
  179.                     value.remove("displays");
  180.                     value.remove("actions");
  181.                     collector.collect(value.toJSONString());
  182.                 }
  183.             }
  184.         });
  186.         //7.提取各个侧输出流数据
  187.         DataStream<String> startDS = pageDS.getSideOutput(startTag);
  188.         DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
  189.         DataStream<String> actionDS = pageDS.getSideOutput(actionTag);
  190.         DataStream<String> errorDS = pageDS.getSideOutput(errorTag);
  192.         //8.将数据打印写入对应的主题
  193.         pageDS.print("Page>>>>>>>>>");
  194.         startDS.print("Start>>>>>>>");
  195.         displayDS.print("Display>>>");
  196.         actionDS.print("Action>>>>>");
  197.         errorDS.print("Error>>>>>>>");
  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";
  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));
  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 图解

17512ce8500c414d8ffbf5c1d8623c56.png

2.4 代码

1**）主程序**

  1. package com.atguigu.app.dwd;
  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;
  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)
  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);
  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.          */
  46.         //2.读取kafka页面日志主题创建流
  48.         String topic= "dwd_traffic_page_log";
  49.        // String topic= "topic_log";
  50.         String groupId= "unique_visitor_detail_211126";
  51.         DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
  53.         //3.过滤掉上一跳页面不为null的数据并将每行数据转换为JSON对象
  54.         SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
  55.             @Override
  56.             public void flatMap(String value, Collector<JSONObject> 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.                 }
  69.             }
  70.         });
  72.         //4.按照mid分组
  73.         KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"));
  75.         //5.使用状态编程实现按照mid的去重
  76.         SingleOutputStreamOperator<JSONObject> uvDS = keyedStream.filter(new RichFilterFunction<JSONObject>() {
  78.             private ValueState<String> lastVisitState;
  80.             @Override
  81.             public void open(Configuration parameters) throws Exception {
  82.                 ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("last-visit", String.class);
  84.                 //设置状态的TTL
  85.                 StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1))
  86.                         .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
  87.                         .build();
  88.                 stateDescriptor.enableTimeToLive(ttlConfig);
  90.                 lastVisitState = getRuntimeContext().getState(stateDescriptor);
  91.             }
  93.             @Override
  94.             public boolean filter(JSONObject value) throws Exception {
  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.                 }
  107.                 return false;
  108.             }
  109.         });
  111.         //6.将数据写入kafka
  112.         String targetTopic = "dwd_traffic_unique_visitor_detail";
  113.         uvDS.print(">>>>>>>>>>");
  114.         uvDS.map(JSONAware::toJSONString)
  115.                 .addSink(MyKafkaUtil.getFlinkKafkaProducer(targetTopic));
  117.         //7.执行
  118.         env.execute();
  119.     }
  120. }

发表评论

表情：
评论列表 （有 0 条评论，203人围观）

还没有评论，来说两句吧...

相关阅读