• Flink实时仓库-DWD层(流量域)模板代码

    简介

    对于DWD层开发起到简介的作用

    工具类

    时间工具类

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

    kafka工具类 

    1. public class KafkaUtil {
    2.     static String BOOTSTRAP_SERVERS = "master:9092, node1:9092, node2:9092";
    3.     static String DEFAULT_TOPIC = "default_topic";
    4.  
    5.     /**
    6.      * 根据主题还有消费者组得到消费者
    7.      * @param topic
    8.      * @param groupId
    9.      * @return
    10.      */
    11.     public static FlinkKafkaConsumer getKafkaConsumer(String topic, String groupId) {
    12.         Properties prop = new Properties();
    13.         prop.setProperty("bootstrap.servers", BOOTSTRAP_SERVERS);
    14.         prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    15.  
    16.         FlinkKafkaConsumer consumer = new FlinkKafkaConsumer<>(topic,
    17.                 //由于默认的解码器，如果字符串为空的时候他会保存，所以自定义一个
    18.                 new KafkaDeserializationSchema() {
    19.                     @Override
    20.                     public boolean isEndOfStream(String nextElement) {
    21.                         return false;
    22.                     }
    23.  
    24.                     @Override
    25.                     public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
    26.                         if(record == null || record.value() == null) {
    27.                             return "";
    28.                         }
    29.                         return new String(record.value());
    30.                     }
    31.  
    32.                     @Override
    33.                     public TypeInformation getProducedType() {
    34.                         return BasicTypeInfo.STRING_TYPE_INFO;
    35.                     }
    36.                 }, prop);
    37.         return consumer;
    38.     }
    39.  
    40.     /**
    41.      * 根据主题得到生产者
    42.      * @param topic
    43.      * @return
    44.      */
    45.     public static FlinkKafkaProducer getKafkaProducer(String topic) {
    46.  
    47.         Properties prop = new Properties();
    48.         prop.setProperty("bootstrap.servers", BOOTSTRAP_SERVERS);
    49.         prop.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60 * 15 * 1000 + "");
    50.         FlinkKafkaProducer producer = new FlinkKafkaProducer(DEFAULT_TOPIC, new KafkaSerializationSchema() {
    51.  
    52.             @Override
    53.             public ProducerRecord<byte[], byte[]> serialize(String jsonStr, @Nullable Long timestamp) {
    54.                 return new ProducerRecord<byte[], byte[]>(topic, jsonStr.getBytes());
    55.             }
    56.         }, prop,
    57.                 FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    58.         return producer;
    59.     }
    60. }

    流量域

    前置知识

    键值状态

    1. public class Status {
    2.     public static void main(String[] args) throws Exception {
    3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    4.         DataStreamSource initData = env.fromElements("a", "a", "b", "c");
    5.  
    6.         //先装换成键值状态才能够使用状态变量
    7.         SingleOutputStreamOperator> mapKeyData = initData.map(new MapFunction>() {
    8.             @Override
    9.             public Tuple2 map(String dataItem) throws Exception {
    10.                 return Tuple2.of(dataItem, 1);
    11.             }
    12.         });
    13.  
    14.         SingleOutputStreamOperator> firstViewDtState = mapKeyData.keyBy(data -> data.f0)
    15.                 .process(new KeyedProcessFunction, Tuple2>() {
    16.                     ValueState firstViewDtState;
    17.  
    18.                     @Override
    19.                     public void open(Configuration param) throws Exception {
    20.                         super.open(param);
    21.                         firstViewDtState = getRuntimeContext().getState(new ValueStateDescriptor(
    22.                                 "firstViewDtState", String.class
    23.                         ));
    24.                     }
    25.  
    26.  
    27.                     @Override
    28.                     public void processElement(Tuple2 value, Context ctx, Collector> out) throws Exception {
    29.                         String keyStatus = firstViewDtState.value();
    30.                         if (keyStatus == null) {
    31.                             //用key的值保存进去
    32.                             firstViewDtState.update(value.f0);
    33.                         } else {
    34.                             System.out.println(value + "  重复来了");
    35.                         }
    36.                     }
    37.                 });
    38.  
    39.         firstViewDtState.print();
    40.  
    41.         env.execute();
    42.     }
    43. }

    结果

    (a,1)  重复来了

    侧输出流

    1. public class OutputTagTest {
    2.     public static void main(String[] args) throws Exception {
    3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    4.         DataStreamSource initData = env.fromElements("a", "a", "b", "c");
    5.  
    6.         // 1 定义侧输出流
    7.         OutputTag a = new OutputTag("a") {
    8.         };
    9.         OutputTag b = new OutputTag("b") {
    10.         };
    11.  
    12.         SingleOutputStreamOperator processData = initData.process(new ProcessFunction() {
    13.             @Override
    14.             public void processElement(String value, Context ctx, Collector out) throws Exception {
    15.                 if (value.equals("a")) {
    16.                     //写到侧输出流a
    17.                     ctx.output(a, value);
    18.                 } else if (value.equals("b")) {
    19.                     //写到侧输出流b
    20.                     ctx.output(b, value);
    21.                 } else {
    22.                     //写出到主流
    23.                     out.collect(value);
    24.                 }
    25.             }
    26.         });
    27.  
    28.         //得到a的侧输出流
    29.         processData.getSideOutput(a).print("a>>");
    30.         //得到b的侧输出流
    31.         processData.getSideOutput(b).print("b>>");
    32.         //主流数据输出
    33.         processData.print("主流>>");
    34.  
    35.         env.execute();
    36.     }
    37. }

    结果

    1. b>>:4> b
    2. a>>:3> a
    3. a>>:2> a
    4. 主流>>:5> c

    日志数据分流

    首先为什么要对于启动日志进行分流？因为可以对于不同的日志类型分别分析，在分析的时候可以减少数据量

    分流实现的程序示例

    1. public class BaseLogApp {
    2.     public static void main(String[] args) throws Exception {
    3.  
    4.         // TODO 1. 初始化环境
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         env.setParallelism(4);
    7.  
    8.         // TODO 2. 启用状态后端
    9.         env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
    10.         env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
    11.         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
    12.         env.getCheckpointConfig().enableExternalizedCheckpoints(
    13.                 CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
    14.         );
    15.         env.setRestartStrategy(RestartStrategies
    16.                 .failureRateRestart(10,
    17.                         Time.of(3L, TimeUnit.DAYS),
    18.                         Time.of(1L, TimeUnit.MINUTES)));
    19.         env.setStateBackend(new HashMapStateBackend());
    20.         env.getCheckpointConfig().setCheckpointStorage("hdfs://master:8020/gmall/ck");
    21.         System.setProperty("HADOOP_USER_NAME", "bigdata");
    22.  
    23.         // TODO 3. 从 Kafka 读取主流数据
    24.         String topic = "topic_logg";
    25.         String groupId = "base_log_consumer";
    26.         DataStreamSource source = env.addSource(KafkaUtil.getKafkaConsumer(topic, groupId));
    27.  
    28.         // TODO 4. 数据清洗，转换结构
    29.         // 4.1 定义错误侧输出流
    30.         OutputTag dirtyStreamTag = new OutputTag("dirtyStream") {
    31.         };
    32.         SingleOutputStreamOperator cleanedStream = source.process(
    33.                 new ProcessFunction() {
    34.                     @Override
    35.                     public void processElement(String jsonStr, Context ctx, Collector out) throws Exception {
    36.                         try {
    37.                             JSONObject jsonObj = JSON.parseObject(jsonStr);
    38.                             out.collect(jsonStr);
    39.                         } catch (Exception e) {
    40.                             ctx.output(dirtyStreamTag, jsonStr);
    41.                         }
    42.                     }
    43.                 }
    44.         );
    45.  
    46.         // 4.2 将脏数据写出到 Kafka 指定主题
    47.         DataStream dirtyStream = cleanedStream.getSideOutput(dirtyStreamTag);
    48.         String dirtyTopic = "dirty_data";
    49.         dirtyStream.addSink(KafkaUtil.getKafkaProducer(dirtyTopic));
    50.  
    51.         // 4.3 转换主流数据结构 jsonStr -> jsonObj
    52.         SingleOutputStreamOperator mappedStream = cleanedStream.map(JSON::parseObject);
    53.  
    54.         // TODO 5. 新老访客状态标记修复
    55.         // 5.1 按照 mid 对数据进行分组
    56.         KeyedStream keyedStream = mappedStream.keyBy(r -> r.getJSONObject("common").getString("mid"));
    57.  
    58.         // 5.2 新老访客状态标记修复
    59.         SingleOutputStreamOperator fixedStream = keyedStream.process(
    60.                 new KeyedProcessFunction() {
    61.  
    62.                     ValueState firstViewDtState;
    63.  
    64.                     @Override
    65.                     public void open(Configuration param) throws Exception {
    66.                         super.open(param);
    67.                         firstViewDtState = getRuntimeContext().getState(new ValueStateDescriptor(
    68.                                 "lastLoginDt", String.class
    69.                         ));
    70.                     }
    71.  
    72.                     @Override
    73.                     public void processElement(JSONObject jsonObj, Context ctx, Collector out) throws Exception {
    74.                         String isNew = jsonObj.getJSONObject("common").getString("is_new");
    75.                         String firstViewDt = firstViewDtState.value();
    76.                         Long ts = jsonObj.getLong("ts");
    77.                         String dt = DateFormatUtil.toDate(ts);
    78.  
    79.                         if ("1".equals(isNew)) {
    80.                             if (firstViewDt == null) {
    81.                                 firstViewDtState.update(dt);
    82.                             } else {
    83.                                 if (!firstViewDt.equals(dt)) {
    84.                                     isNew = "0";
    85.                                     jsonObj.getJSONObject("common").put("is_new", isNew);
    86.                                 }
    87.                             }
    88.                         } else {
    89.                             if (firstViewDt == null) {
    90.                                 // 将首次访问日期置为昨日
    91.                                 String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
    92.                                 firstViewDtState.update(yesterday);
    93.                             }
    94.                         }
    95.  
    96.                         out.collect(jsonObj);
    97.                     }
    98.                 }
    99.         );
    100.  
    101.         // TODO 6. 分流
    102.         // 6.1 定义启动、曝光、动作、错误侧输出流
    103.         OutputTag startTag = new OutputTag("startTag") {
    104.         };
    105.         OutputTag displayTag = new OutputTag("displayTag") {
    106.         };
    107.         OutputTag actionTag = new OutputTag("actionTag") {
    108.         };
    109.         OutputTag errorTag = new OutputTag("errorTag") {
    110.         };
    111.  
    112.         // 6.2 分流
    113.         SingleOutputStreamOperator separatedStream = fixedStream.process(
    114.                 new ProcessFunction() {
    115.                     @Override
    116.                     public void processElement(JSONObject jsonObj, Context context, Collector out) throws Exception {
    117.  
    118.                         // 6.2.1 收集错误数据
    119.                         JSONObject error = jsonObj.getJSONObject("err");
    120.                         if (error != null) {
    121.                             context.output(errorTag, jsonObj.toJSONString());
    122.                         }
    123.  
    124.                         // 剔除 "err" 字段
    125.                         jsonObj.remove("err");
    126.  
    127.                         // 6.2.2 收集启动数据
    128.                         JSONObject start = jsonObj.getJSONObject("start");
    129.                         if (start != null) {
    130.                             context.output(startTag, jsonObj.toJSONString());
    131.                         } else {
    132.                             // 获取 "page" 字段
    133.                             JSONObject page = jsonObj.getJSONObject("page");
    134.                             // 获取 "common" 字段
    135.                             JSONObject common = jsonObj.getJSONObject("common");
    136.                             // 获取 "ts"
    137.                             Long ts = jsonObj.getLong("ts");
    138.  
    139.                             // 6.2.3 收集曝光数据
    140.                             JSONArray displays = jsonObj.getJSONArray("displays");
    141.                             if (displays != null) {
    142.                                 for (int i = 0; i < displays.size(); i++) {
    143.                                     JSONObject display = displays.getJSONObject(i);
    144.                                     JSONObject displayObj = new JSONObject();
    145.                                     displayObj.put("display", display);
    146.                                     displayObj.put("common", common);
    147.                                     displayObj.put("page", page);
    148.                                     displayObj.put("ts", ts);
    149.                                     context.output(displayTag, displayObj.toJSONString());
    150.                                 }
    151.                             }
    152.  
    153.                             // 6.2.4 收集动作数据
    154.                             JSONArray actions = jsonObj.getJSONArray("actions");
    155.                             if (actions != null) {
    156.                                 for (int i = 0; i < actions.size(); i++) {
    157.                                     JSONObject action = actions.getJSONObject(i);
    158.                                     JSONObject actionObj = new JSONObject();
    159.                                     actionObj.put("action", action);
    160.                                     actionObj.put("common", common);
    161.                                     actionObj.put("page", page);
    162.                                     actionObj.put("ts", ts);
    163.                                     context.output(actionTag, actionObj.toJSONString());
    164.                                 }
    165.                             }
    166.  
    167.                             // 6.2.5 收集页面数据
    168.                             jsonObj.remove("displays");
    169.                             jsonObj.remove("actions");
    170.                             out.collect(jsonObj.toJSONString());
    171.                         }
    172.  
    173.                     }
    174.                 }
    175.         );
    176.  
    177. //         打印主流和各侧输出流查看分流效果
    178.         separatedStream.print("page>>>");
    179.         separatedStream.getSideOutput(startTag).print("start!!!");
    180.         separatedStream.getSideOutput(displayTag).print("display@@@");
    181.         separatedStream.getSideOutput(actionTag).print("action###");
    182.         separatedStream.getSideOutput(errorTag).print("error$$$");
    183.  
    184.         // TODO 7. 将数据输出到 Kafka 的不同主题
    185. //        // 7.1 提取各侧输出流
    186. //        DataStream startDS = separatedStream.getSideOutput(startTag);
    187. //        DataStream displayDS = separatedStream.getSideOutput(displayTag);
    188. //        DataStream actionDS = separatedStream.getSideOutput(actionTag);
    189. //        DataStream errorDS = separatedStream.getSideOutput(errorTag);
    190. //
    191. //        // 7.2 定义不同日志输出到 Kafka 的主题名称
    192. //        String page_topic = "dwd_traffic_page_log";
    193. //        String start_topic = "dwd_traffic_start_log";
    194. //        String display_topic = "dwd_traffic_display_log";
    195. //        String action_topic = "dwd_traffic_action_log";
    196. //        String error_topic = "dwd_traffic_error_log";
    197. //
    198. //        separatedStream.addSink(KafkaUtil.getKafkaProducer(page_topic));
    199. //        startDS.addSink(KafkaUtil.getKafkaProducer(start_topic));
    200. //        displayDS.addSink(KafkaUtil.getKafkaProducer(display_topic));
    201. //        actionDS.addSink(KafkaUtil.getKafkaProducer(action_topic));
    202. //        errorDS.addSink(KafkaUtil.getKafkaProducer(error_topic));
    203.  
    204.         env.execute();
    205.     }
    206. }

    初始日志数据

    1. {
    2.     "actions": [
    3.         {
    4.             "action_id": "get_coupon",
    5.             "item": "3",
    6.             "item_type": "coupon_id",
    7.             "ts": 1592134620882
    8.         }
    9.     ],
    10.     "common": {
    11.         "ar": "110000",
    12.         "ba": "Oneplus",
    13.         "ch": "oppo",
    14.         "is_new": "0",
    15.         "md": "Oneplus 7",
    16.         "mid": "mid_232163",
    17.         "os": "Android 10.0",
    18.         "uid": "898",
    19.         "vc": "v2.1.134"
    20.     },
    21.     "displays": [
    22.         {
    23.             "display_type": "query",
    24.             "item": "18",
    25.             "item_type": "sku_id",
    26.             "order": 1,
    27.             "pos_id": 2
    28.         },
    29.         {
    30.             "display_type": "promotion",
    31.             "item": "13",
    32.             "item_type": "sku_id",
    33.             "order": 2,
    34.             "pos_id": 4
    35.         },
    36.         {
    37.             "display_type": "query",
    38.             "item": "29",
    39.             "item_type": "sku_id",
    40.             "order": 3,
    41.             "pos_id": 2
    42.         },
    43.         {
    44.             "display_type": "query",
    45.             "item": "7",
    46.             "item_type": "sku_id",
    47.             "order": 4,
    48.             "pos_id": 4
    49.         },
    50.         {
    51.             "display_type": "query",
    52.             "item": "19",
    53.             "item_type": "sku_id",
    54.             "order": 5,
    55.             "pos_id": 4
    56.         },
    57.         {
    58.             "display_type": "promotion",
    59.             "item": "22",
    60.             "item_type": "sku_id",
    61.             "order": 6,
    62.             "pos_id": 4
    63.         },
    64.         {
    65.             "display_type": "query",
    66.             "item": "25",
    67.             "item_type": "sku_id",
    68.             "order": 7,
    69.             "pos_id": 5
    70.         }
    71.     ],
    72.     "page": {
    73.         "during_time": 11764,
    74.         "item": "31",
    75.         "item_type": "sku_id",
    76.         "last_page_id": "good_list",
    77.         "page_id": "good_detail",
    78.         "source_type": "query"
    79.     },
    80.     "ts": 1592134615000
    81. }

    分流后数据

    主流(启动日志数据)

    1. {
    2.     "common": {
    3.         "ar": "310000",
    4.         "uid": "780",
    5.         "os": "Android 11.0",
    6.         "ch": "oppo",
    7.         "is_new": "1",
    8.         "md": "Huawei P30",
    9.         "mid": "mid_619118",
    10.         "vc": "v2.1.134",
    11.         "ba": "Huawei"
    12.     },
    13.     "page": {
    14.         "page_id": "payment",
    15.         "item": "33,25",
    16.         "during_time": 5439,
    17.         "item_type": "sku_ids",
    18.         "last_page_id": "trade"
    19.     },
    20.     "ts": 1592134614000
    21. }

    曝光日志数据

    1. {
    2.     "common": {
    3.         "ar": "110000",
    4.         "uid": "914",
    5.         "os": "iOS 13.3.1",
    6.         "ch": "Appstore",
    7.         "is_new": "0",
    8.         "md": "iPhone Xs Max",
    9.         "mid": "mid_319090",
    10.         "vc": "v2.1.134",
    11.         "ba": "iPhone"
    12.     },
    13.     "display": {
    14.         "display_type": "query",
    15.         "item": "8",
    16.         "item_type": "sku_id",
    17.         "pos_id": 1,
    18.         "order": 7
    19.     },
    20.     "page": {
    21.         "page_id": "home",
    22.         "during_time": 4328
    23.     },
    24.     "ts": 1592134610000
    25. }

    行为日志数据

    1. {
    2.     "common": {
    3.         "ar": "230000",
    4.         "uid": "257",
    5.         "os": "Android 11.0",
    6.         "ch": "vivo",
    7.         "is_new": "0",
    8.         "md": "Xiaomi 9",
    9.         "mid": "mid_227516",
    10.         "vc": "v2.0.1",
    11.         "ba": "Xiaomi"
    12.     },
    13.     "action": {
    14.         "item": "35",
    15.         "action_id": "cart_minus_num",
    16.         "item_type": "sku_id",
    17.         "ts": 1592134612791
    18.     },
    19.     "page": {
    20.         "page_id": "cart",
    21.         "during_time": 3583,
    22.         "last_page_id": "good_detail"
    23.     },
    24.     "ts": 1592134611000
    25. }

    错误日志数据

    1. {
    2.     "common": {
    3.         "ar": "110000",
    4.         "uid": "780",
    5.         "os": "Android 11.0",
    6.         "ch": "huawei",
    7.         "is_new": "1",
    8.         "md": "Xiaomi 9",
    9.         "mid": "mid_503805",
    10.         "vc": "v2.1.134",
    11.         "ba": "Xiaomi"
    12.     },
    13.     "err": {
    14.         "msg": " Exception in thread \\  java.net.SocketTimeoutException\\n \\tat com.atgugu.gmall2020.mock.bean.log.AppError.main(AppError.java:xxxxxx)",
    15.         "error_code": 1245
    16.     },
    17.     "page": {
    18.         "page_id": "home",
    19.         "during_time": 17642
    20.     },
    21.     "displays": [
    22.         {
    23.             "display_type": "activity",
    24.             "item": "1",
    25.             "item_type": "activity_id",
    26.             "pos_id": 2,
    27.             "order": 1
    28.         },
    29.         {
    30.             "display_type": "query",
    31.             "item": "2",
    32.             "item_type": "sku_id",
    33.             "pos_id": 5,
    34.             "order": 2
    35.         },
    36.         {
    37.             "display_type": "query",
    38.             "item": "6",
    39.             "item_type": "sku_id",
    40.             "pos_id": 2,
    41.             "order": 3
    42.         },
    43.         {
    44.             "display_type": "query",
    45.             "item": "8",
    46.             "item_type": "sku_id",
    47.             "pos_id": 4,
    48.             "order": 4
    49.         },
    50.         {
    51.             "display_type": "query",
    52.             "item": "6",
    53.             "item_type": "sku_id",
    54.             "pos_id": 3,
    55.             "order": 5
    56.         },
    57.         {
    58.             "display_type": "promotion",
    59.             "item": "29",
    60.             "item_type": "sku_id",
    61.             "pos_id": 3,
    62.             "order": 6
    63.         }
    64.     ],
    65.     "ts": 1592134611000
    66. }

    最终的结果

    独立访客实时表

    前提知识

    状态TTL

    1. public class StateTTl {
    2.     public static void main(String[] args) throws Exception {
    3.         // 环境准备
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.         env.setParallelism(1);
    6.  
    7.         //初始化数据
    8.         DataStreamSource initData = env.fromElements("a", "a", "b", "c", "b");
    9.  
    10.         SingleOutputStreamOperator> targetData = initData.map(new MapFunction>() {
    11.             @Override
    12.             public Tuple2 map(String value) throws Exception {
    13.                 if (value.equals("a")) {
    14.                     Thread.sleep(2000);
    15.                 }
    16.                 return Tuple2.of(value,1);
    17.             }
    18.         });
    19.  
    20.         //测试状态过期
    21.         SingleOutputStreamOperator> testTTL = targetData.keyBy(data -> data.f0)
    22.                 .process(new KeyedProcessFunction, Tuple2>() {
    23.                     private ValueState lastVisitDt;
    24.  
    25.                     @Override
    26.                     public void open(Configuration paramenters) throws Exception {
    27.                         super.open(paramenters);
    28.                         ValueStateDescriptor valueStateDescriptor =
    29.                                 new ValueStateDescriptor<>("testTTL", String.class);
    30.                         valueStateDescriptor.enableTimeToLive(
    31.                                 StateTtlConfig
    32.                                         .newBuilder(Time.seconds(1L))
    33.                                         // 设置在创建和更新状态时更新存活时间
    34.                                         .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    35.                                         .build()
    36.                         );
    37.                         //对于keyStatus设置ttl
    38.                         lastVisitDt = getRuntimeContext().getState(valueStateDescriptor);
    39.                     }
    40.  
    41.                     @Override
    42.                     public void processElement(Tuple2 value, KeyedProcessFunction, Tuple2>.Context ctx, Collector> out) throws Exception {
    43.                         String valueState = lastVisitDt.value();
    44.                         if (valueState == null) {
    45.                             System.out.println(value);
    46.                             lastVisitDt.update(value.f0);
    47.                         } else {
    48.                             System.out.println(value.f0 + " 有对应的状态了");
    49.                         }
    50.                     }
    51.                 });
    52.  
    53.         testTTL.print();
    54.  
    55.  
    56.         env.execute();
    57.     }
    58. }

    输出结果

    1. (a,1)
    2. (a,1)
    3. (b,1)
    4. (c,1)
    5. b 有对应的状态了

    结果可以看出第一个a经过一秒以后删除了

    代码实现

    1. public class DwdTrafficUniqueVisitorDetail {
    2.     public static void main(String[] args) throws Exception {
    3.  
    4.         // TODO 1. 环境准备
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         env.setParallelism(4);
    7.  
    8.         // TODO 2. 状态后端设置
    9.         env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
    10.         env.getCheckpointConfig().setCheckpointTimeout(30 * 1000L);
    11.         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
    12.         env.getCheckpointConfig().enableExternalizedCheckpoints(
    13.                 CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
    14.         );
    15.         env.setRestartStrategy(RestartStrategies.failureRateRestart(
    16.                 3, Time.days(1), Time.minutes(1)
    17.         ));
    18.         env.setStateBackend(new HashMapStateBackend());
    19.         env.getCheckpointConfig().setCheckpointStorage(
    20.                 "hdfs://master:8020/ck"
    21.         );
    22.         System.setProperty("HADOOP_USER_NAME", "bigdata");
    23.  
    24.         // TODO 3. 从 kafka dwd_traffic_page_log 主题读取日志数据，封装为流
    25.         String topic = "dwd_traffic_page_log";
    26.         String groupId = "dwd_traffic_user_jump_detail";
    27.         FlinkKafkaConsumer kafkaConsumer = KafkaUtil.getKafkaConsumer(topic, groupId);
    28.         DataStreamSource pageLog = env.addSource(kafkaConsumer);
    29.  
    30.         // TODO 4. 转换结构
    31.         SingleOutputStreamOperator mappedStream = pageLog.map(JSON::parseObject);
    32.  
    33.         // TODO 5. 过滤 last_page_id 不为 null 的数据
    34.         SingleOutputStreamOperator firstPageStream = mappedStream.filter(
    35.                 jsonObj -> jsonObj
    36.                         .getJSONObject("page")
    37.                         .getString("last_page_id") == null
    38.         );
    39.  
    40.         // TODO 6. 按照 mid 分组
    41.         KeyedStream keyedStream = firstPageStream
    42.                 .keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
    43.  
    44.         // TODO 7. 通过 Flink 状态编程过滤独立访客记录
    45.         SingleOutputStreamOperator filteredStream = keyedStream.filter(
    46.                 new RichFilterFunction() {
    47.  
    48.                     private ValueState lastVisitDt;
    49.  
    50.                     @Override
    51.                     public void open(Configuration paramenters) throws Exception {
    52.                         super.open(paramenters);
    53.                         ValueStateDescriptor valueStateDescriptor =
    54.                                 new ValueStateDescriptor<>("last_visit_dt", String.class);
    55.                         valueStateDescriptor.enableTimeToLive(
    56.                                 StateTtlConfig
    57.                                         .newBuilder(Time.days(1L))
    58.                                         // 设置在创建和更新状态时更新存活时间
    59.                                         .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    60.                                         .build()
    61.                         );
    62.                         lastVisitDt = getRuntimeContext().getState(valueStateDescriptor);
    63.                     }
    64.  
    65.                     @Override
    66.                     public boolean filter(JSONObject jsonObj) throws Exception {
    67.                         String visitDt = DateFormatUtil.toDate(jsonObj.getLong("ts"));
    68.                         String lastDt = lastVisitDt.value();
    69.                         if (lastDt == null || !lastDt.equals(visitDt)) {
    70.                             lastVisitDt.update(visitDt);
    71.                             return true;
    72.                         }
    73.                         return false;
    74.                     }
    75.                 }
    76.         );
    77.  
    78.         // TODO 8. 将独立访客数据写入
    79.         // Kafka dwd_traffic_unique_visitor_detail 主题
    80.         String targetTopic = "dwd_traffic_unique_visitor_detail";
    81.         FlinkKafkaProducer kafkaProducer = KafkaUtil.getKafkaProducer(targetTopic);
    82.         filteredStream.map(JSONAware::toJSONString).addSink(kafkaProducer);
    83.  
    84.         // TODO 9. 启动任务
    85.         env.execute();
    86.     }
    87. }

    结果

    1. {
    2.     "common": {
    3.         "ar": "310000",
    4.         "uid": "201",
    5.         "os": "Android 11.0",
    6.         "ch": "vivo",
    7.         "is_new": "1",
    8.         "md": "Xiaomi Mix2 ",
    9.         "mid": "mid_994205",
    10.         "vc": "v2.0.1",
    11.         "ba": "Xiaomi"
    12.     },
    13.     "page": {
    14.         "page_id": "home",
    15.         "during_time": 19868
    16.     },
    17.     "ts": 1592133292000
    18. }

    跳出事务事实表

    前提知识

    要保证数据是有序的首先使用的是水位线，然后使用窗口的操作才能使数据有序

    FlinkCEP的使用

    1. public class CEPTest {
    2.     public static void main(String[] args) throws Exception {
    3.         //TODO 1得到执行环境
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.         //TODO 注意在多个分区的时候      因为并行度多个的话 ，watermark还是不会被提升的，还是触发不了计算
    6.         env.setParallelism(1);
    7.         //TODO 2得到网络端口的数据 源
    8.         DataStreamSource socketTextStream = env.socketTextStream("master", 9998);
    9.         //TODO 3假设我们得到的数据是a 1,第一个是数据第二个是秒
    10.         SingleOutputStreamOperator> mappedStream = socketTextStream.map(new MapFunction>() {
    11.             @Override
    12.             public Tuple2 map(String value) throws Exception {
    13.                 String[] s = value.split(" ");
    14.                 return Tuple2.of(s[0], Integer.parseInt(s[1])*1000L);
    15.             }
    16.         });
    17.  
    18.         //TODO 4为了保证数据的有序就得使用水位线
    19.         SingleOutputStreamOperator> withWatermarkStream = mappedStream.assignTimestampsAndWatermarks(
    20.                 WatermarkStrategy
    21.                         //设置延迟0秒
    22.                         .>forBoundedOutOfOrderness(Duration.ofSeconds(0))
    23.                         .withTimestampAssigner(
    24.                                 new SerializableTimestampAssigner>() {
    25.                                     @Override
    26.                                     public long extractTimestamp(Tuple2 initData, long recordTimestamp) {
    27. //                                        System.out.println(initData);
    28.                                         return initData.f1;
    29.                                     }
    30.                                 }
    31.                         )
    32.         );
    33.  
    34.         //TODO 5 分组以后对于每一个组进行分组开窗
    35.         KeyedStream, String> keyedStream = withWatermarkStream.keyBy(data -> {
    36.             return data.f0;
    37.         });
    38.  
    39. //        keyedStream.print("keyedStream");
    40.  
    41.         //TODO 6这里主要是使用CEP匹配到自己想要的数据，我们这里想要的是a 和 b在10秒内是连续的,否则就是超时了
    42.         Pattern, Tuple2> pattern = Pattern.>begin("first").where(
    43.                 new SimpleCondition>() {
    44.  
    45.                     @Override
    46.                     public boolean filter(Tuple2 firstData) throws Exception {
    47.                         return firstData.f0.equals("a");
    48.                     }
    49.                 }
    50.         ).next("second").where(
    51.                 new SimpleCondition>() {
    52.                     @Override
    53.                     public boolean filter(Tuple2 secondData) throws Exception {
    54.                         return secondData.f0.equals("a");
    55.                     }
    56.                 }
    57.                 // 上文调用了同名 Time 类，此处需要使用全类名
    58.         ).within(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L));
    59.  
    60.         PatternStream> patternStream = CEP.pattern(keyedStream, pattern);
    61.  
    62.         //超时侧输出流
    63.         OutputTag> timeoutTag = new OutputTag>("timeoutTag") {
    64.         };
    65.         SingleOutputStreamOperator> flatSelectStream = patternStream.flatSelect(
    66.                 timeoutTag,
    67.                 new PatternFlatTimeoutFunction, Tuple2>() {
    68.                     @Override
    69.                     public void timeout(Map>> pattern, long timeoutTimestamp, Collector> out) throws Exception {
    70.                         //得到超时的数据
    71.                         Tuple2 first = pattern.get("first").get(0);
    72.                         out.collect(first);
    73.                     }
    74.                 },
    75.                 new PatternFlatSelectFunction, Tuple2>() {
    76.                     @Override
    77.                     public void flatSelect(Map>> pattern, Collector> out) throws Exception {
    78.                         //得到了匹配到的数据
    79.                         Tuple2 second = pattern.get("second").get(0);
    80.                         out.collect(second);
    81.                     }
    82.                 }
    83.         );
    84.  
    85.         DataStream> timeOutDStream = flatSelectStream.getSideOutput(timeoutTag);
    86.         //打印超时的数据
    87.         timeOutDStream.print("timeOut");
    88.         //打印正常匹配到的数据
    89.         flatSelectStream.print("flatSelectStream");
    90.  
    91.         env.execute();
    92.     }
    93. }

    测试数据

    1. [bigdata@master createdata]$ nc -lk 9998
    2. a 2
    3. a 3
    4. a 15

    得到结果

    1. flatSelectStream> (a,3000)
    2. timeOut> (a,3000)

    结论在a 2,a 3的时候还没有关闭10秒的窗口，输入15的时候关闭窗口，由于a 2,a 3然后a 3是符合10秒内的数据的所以输出，然后是输入a 15,那么前面的a 3就是超时的数据

    Union

    1. public class UnionTest {
    2.     public static void main(String[] args) throws Exception {
    3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    4.         DataStreamSource dataStream1 = env.fromElements(1, 2, 3);
    5.         DataStreamSource dataStream2 = env.fromElements(4, 5, 6);
    6.  
    7.         dataStream1.union(dataStream2)
    8.                         .print();
    9.         env.execute();
    10.     }
    11. }

    输出

    1. 2> 5
    2. 3> 6
    3. 1> 4
    4. 11> 3
    5. 9> 1
    6. 10> 2

    实战代码

    1. public class DwdTrafficUserJumpDetail {
    2.     public static void main(String[] args) throws Exception {
    3.  
    4.         // TODO 1. 环境准备
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         env.setParallelism(4);
    7.  
    8.         // TODO 2. 状态后端设置
    9.         env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
    10.         env.getCheckpointConfig().setCheckpointTimeout(30 * 1000L);
    11.         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
    12.         env.getCheckpointConfig().enableExternalizedCheckpoints(
    13.                 CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
    14.         );
    15.         env.setRestartStrategy(RestartStrategies.failureRateRestart(
    16.                 3, Time.days(1), Time.minutes(1)
    17.         ));
    18.         env.setStateBackend(new HashMapStateBackend());
    19.         env.getCheckpointConfig().setCheckpointStorage(
    20.                 "hdfs://master:8020/ck"
    21.         );
    22.         System.setProperty("HADOOP_USER_NAME", "bigdata");
    23.  
    24.         // TODO 3. 从 kafka dwd_traffic_page_log 主题读取日志数据，封装为流
    25.         String topic = "dwd_traffic_page_log";
    26.         String groupId = "dwd_traffic_user_jump_detail";
    27.         FlinkKafkaConsumer kafkaConsumer = KafkaUtil.getKafkaConsumer(topic, groupId);
    28.         DataStreamSource pageLog = env.addSource(kafkaConsumer);
    29.  
    30.         // 测试数据
    31.         /*DataStream kafkaDS = env
    32.             .fromElements(
    33.                 "{\"common\":{\"mid\":\"101\"},\"page\":{\"page_id\":\"home\"},\"ts\":10000} ",
    34.                 "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"home\"},\"ts\":12000}",
    35.                 "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
    36.                     "\"home\"},\"ts\":15000} ",
    37.                 "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
    38.                     "\"detail\"},\"ts\":30000} "
    39.             );*/
    40.  
    41.         // TODO 4. 转换结构
    42.         SingleOutputStreamOperator mappedStream = pageLog.map(JSON::parseObject);
    43.  
    44.         // TODO 5. 设置水位线，用于用户跳出统计
    45.         SingleOutputStreamOperator withWatermarkStream = mappedStream.assignTimestampsAndWatermarks(
    46.                 WatermarkStrategy
    47.                         .forMonotonousTimestamps()
    48.                         .withTimestampAssigner(
    49.                                 new SerializableTimestampAssigner() {
    50.                                     @Override
    51.                                     public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
    52.                                         return jsonObj.getLong("ts");
    53.                                     }
    54.                                 }
    55.                         )
    56.         );
    57.  
    58.         // TODO 6. 按照 mid 分组
    59.         KeyedStream keyedStream = withWatermarkStream.keyBy(jsonOjb -> jsonOjb.getJSONObject("common").getString("mid"));
    60.  
    61.         // TODO 7. 定义 CEP 匹配规则
    62.         Pattern pattern = Pattern.begin("first").where(
    63.                 new SimpleCondition() {
    64.  
    65.                     @Override
    66.                     public boolean filter(JSONObject jsonObj) throws Exception {
    67.                         String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
    68.                         return lastPageId == null;
    69.                     }
    70.                 }
    71.         ).next("second").where(
    72.                 new SimpleCondition() {
    73.                     @Override
    74.                     public boolean filter(JSONObject jsonObj) throws Exception {
    75.                         String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
    76.                         return lastPageId == null;
    77.                     }
    78.                 }
    79.                 // 上文调用了同名 Time 类，此处需要使用全类名
    80.         ).within(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L));
    81.  
    82.         // TODO 8. 把 Pattern 应用到流上
    83.         PatternStream patternStream = CEP.pattern(keyedStream, pattern);
    84.  
    85.         // TODO 9. 提取匹配上的事件以及超时事件
    86.         OutputTag timeoutTag = new OutputTag("timeoutTag") {
    87.         };
    88.         SingleOutputStreamOperator flatSelectStream = patternStream.flatSelect(
    89.                 timeoutTag,
    90.                 new PatternFlatTimeoutFunction() {
    91.                     @Override
    92.                     public void timeout(Map> pattern, long timeoutTimestamp, Collector out) throws Exception {
    93.                         JSONObject element = pattern.get("first").get(0);
    94.                         out.collect(element);
    95.  
    96.                     }
    97.                 },
    98.                 new PatternFlatSelectFunction() {
    99.                     @Override
    100.                     public void flatSelect(Map> pattern, Collector out) throws Exception {
    101.                         JSONObject element = pattern.get("first").get(0);
    102.                         out.collect(element);
    103.                     }
    104.                 }
    105.         );
    106.  
    107.         DataStream timeOutDStream = flatSelectStream.getSideOutput(timeoutTag);
    108.  
    109.         // TODO 11. 合并两个流并将数据写出到 Kafka
    110.         DataStream unionDStream = flatSelectStream.union(timeOutDStream);
    111.         String targetTopic = "dwd_traffic_user_jump_detail";
    112.         FlinkKafkaProducer kafkaProducer = KafkaUtil.getKafkaProducer(targetTopic);
    113.         unionDStream .map(JSONAware::toJSONString)
    114.                 .addSink(kafkaProducer);
    115.  
    116.         env.execute();
    117.     }
    118. }

  • 相关阅读:
    优化软件系统，解决死锁问题，提升稳定性与性能 redis排队下单
    RDD的分区、依赖关系、机制
    win10下使用composer安装依赖ssl报错处理
    【环境搭建】Apache ZooKeeper 3.8.4 Stable
    Java多线程：CopyOnWriteArrayList 实现原理
    Powershell脚本自动备份dhcp数据库
    【Labivew】G语言
    Xss跨站脚本攻击
    嵌入式FreeRTOS学习十一，深入理解任务调度机制
    NCTF-2019-Crypto部分 复现
  • 原文地址：https://blog.csdn.net/S1124654/article/details/125878285