• 尚硅谷大数据项目《在线教育之实时数仓》笔记006

    视频地址：尚硅谷大数据项目《在线教育之实时数仓》_哔哩哔哩_bilibili

    目录

    第9章 数仓开发之DWD层

    P041

    P042

    P043

    P044

    P045

    P046

    P047

    P048

    P049

    P050

    P051

    P052

    第9章 数仓开发之DWD层

    P041

    9.3 流量域用户跳出事务事实表

    P042

    DwdTrafficUserJumpDetail

    // TODO 1 创建环境设置状态后端

    // TODO 2 从kafka的page主题读取数据

    // TODO 3 过滤加转换数据

    // TODO 4 添加水位线

    // TODO 5 按照mid分组

    P043

    1. package com.atguigu.edu.realtime.app.dwd.log;
    2.  
    3. import com.alibaba.fastjson.JSON;
    4. import com.alibaba.fastjson.JSONObject;
    5. import com.atguigu.edu.realtime.util.EnvUtil;
    6. import com.atguigu.edu.realtime.util.KafkaUtil;
    7. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
    8. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    9. import org.apache.flink.api.common.functions.FlatMapFunction;
    10. import org.apache.flink.api.java.functions.KeySelector;
    11. import org.apache.flink.cep.CEP;
    12. import org.apache.flink.cep.PatternFlatSelectFunction;
    13. import org.apache.flink.cep.PatternFlatTimeoutFunction;
    14. import org.apache.flink.cep.PatternStream;
    15. import org.apache.flink.cep.pattern.Pattern;
    16. import org.apache.flink.cep.pattern.conditions.IterativeCondition;
    17. import org.apache.flink.streaming.api.datastream.*;
    18. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    19. import org.apache.flink.streaming.api.windowing.time.Time;
    20. import org.apache.flink.util.Collector;
    21. import org.apache.flink.util.OutputTag;
    22.  
    23. import java.util.List;
    24. import java.util.Map;
    25.  
    26. /**
    27.  * @author yhm
    28.  * @create 2023-04-21 17:54
    29.  */
    30. public class DwdTrafficUserJumpDetail {
    31.     public static void main(String[] args) throws Exception {
    32.         // TODO 1 创建环境设置状态后端
    33.         StreamExecutionEnvironment env = EnvUtil.getExecutionEnvironment(4);
    34.  
    35.         // TODO 2 从kafka的page主题读取数据
    36.         String topicName = "dwd_traffic_page_log";
    37.         DataStreamSource logDS = env.fromSource(KafkaUtil.getKafkaConsumer(topicName, "dwd_traffic_user_jump_detail"), WatermarkStrategy.noWatermarks(), "user_jump_source");
    38.  
    39.         // 测试数据
    40.         DataStream kafkaDS = env
    41.                 .fromElements(
    42.                         "{\"common\":{\"mid\":\"101\"},\"page\":{\"page_id\":\"home\"},\"ts\":10000} ",
    43.                         "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"home\"},\"ts\":12000}",
    44.                         "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\"},\"ts\":15000} ",
    45.                         "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
    46.                                 "\"detail\"},\"ts\":30000} "
    47.                 );
    48.  
    49.         // TODO 3 过滤加转换数据
    50.         SingleOutputStreamOperator jsonObjStream = kafkaDS.flatMap(new FlatMapFunction() {
    51.             @Override
    52.             public void flatMap(String value, Collector out) throws Exception {
    53.                 try {
    54.                     JSONObject jsonObject = JSON.parseObject(value);
    55.                     out.collect(jsonObject);
    56.                 } catch (Exception e) {
    57.                     e.printStackTrace();
    58.                 }
    59.             }
    60.         });
    61.  
    62.         // TODO 4 添加水位线
    63.         SingleOutputStreamOperator withWatermarkStream = jsonObjStream.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps()
    64.                 .withTimestampAssigner(new SerializableTimestampAssigner() {
    65.                     @Override
    66.                     public long extractTimestamp(JSONObject element, long recordTimestamp) {
    67.                         return element.getLong("ts");
    68.                     }
    69.                 }));
    70.  
    71.         // TODO 5 按照mid分组
    72.         KeyedStream keyedStream = withWatermarkStream.keyBy(new KeySelector() {
    73.             @Override
    74.             public String getKey(JSONObject jsonObject) throws Exception {
    75.                 return jsonObject.getJSONObject("common").getString("mid");
    76.             }
    77.         });
    78.  
    79.         // TODO 6 定义cep匹配规则
    80.         Pattern pattern = Pattern.begin("first").where(new IterativeCondition() {
    81.             @Override
    82.             public boolean filter(JSONObject jsonObject, Context ctx) throws Exception {
    83.                 // 一个会话的开头   ->   last_page_id 为空
    84.                 String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
    85.                 return lastPageId == null;
    86.             }
    87.         }).next("second").where(new IterativeCondition() {
    88.             @Override
    89.             public boolean filter(JSONObject jsonObject, Context ctx) throws Exception {
    90.                 // 满足匹配的条件
    91.                 // 紧密相连，又一个会话的开头
    92.                 String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
    93.                 return lastPageId == null;
    94.             }
    95.         }).within(Time.seconds(10L));
    96.  
    97.         // TODO 7 将CEP作用到流上
    98.         PatternStream patternStream = CEP.pattern(keyedStream, pattern);
    99.  
    100.         // TODO 8 提取匹配数据和超时数据
    101.         OutputTag timeoutTag = new OutputTag("timeoutTag") {
    102.         };
    103.         SingleOutputStreamOperator flatSelectStream = patternStream.flatSelect(timeoutTag, new PatternFlatTimeoutFunction() {
    104.             @Override
    105.             public void timeout(Map> pattern, long timeoutTimestamp, Collector out) throws Exception {
    106.                 JSONObject first = pattern.get("first").get(0);
    107.                 out.collect(first.toJSONString());
    108.             }
    109.         }, new PatternFlatSelectFunction() {
    110.             @Override
    111.             public void flatSelect(Map> pattern, Collector out) throws Exception {
    112.                 JSONObject first = pattern.get("first").get(0);
    113.                 out.collect(first.toJSONString());
    114.             }
    115.         });
    116.  
    117.         SideOutputDataStream timeoutStream = flatSelectStream.getSideOutput(timeoutTag);
    118.  
    119.         // TODO 9 合并数据写出到kafka
    120.         DataStream unionStream = flatSelectStream.union(timeoutStream);
    121.         String targetTopic = "dwd_traffic_user_jump_detail";
    122.         unionStream.sinkTo(KafkaUtil.getKafkaProducer(targetTopic, "user_jump_trans"));
    123.  
    124.         // TODO 10 执行任务
    125.         env.execute();
    126.     }
    127. }

    P044

    超时数据

    P045

    9.4 学习域播放事务事实表

    P046

    DwdLearnPlay、DwdLearnPlayBean

    //TODO 1 创建环境设置状态后端

    //TODO 2 读取kafka播放日志数据

    //TODO 3 清洗转换

    //TODO 4 添加水位线

    P047

    1. package com.atguigu.edu.realtime.app.dwd.log;
    2.  
    3. import com.alibaba.fastjson.JSON;
    4. import com.alibaba.fastjson.JSONObject;
    5. import com.atguigu.edu.realtime.bean.DwdLearnPlayBean;
    6. import com.atguigu.edu.realtime.util.EnvUtil;
    7. import com.atguigu.edu.realtime.util.KafkaUtil;
    8. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
    9. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    10. import org.apache.flink.api.common.functions.FlatMapFunction;
    11. import org.apache.flink.api.common.functions.ReduceFunction;
    12. import org.apache.flink.api.java.functions.KeySelector;
    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.datastream.WindowedStream;
    17. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    18. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
    19. import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
    20. import org.apache.flink.streaming.api.windowing.time.Time;
    21. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    22. import org.apache.flink.util.Collector;
    23.  
    24. import java.time.Duration;
    25.  
    26. /**
    27.  * @author yhm
    28.  * @create 2023-04-23 14:21
    29.  */
    30. public class DwdLearnPlay {
    31.     public static void main(String[] args) throws Exception {
    32.         //TODO 1 创建环境设置状态后端
    33.         StreamExecutionEnvironment env = EnvUtil.getExecutionEnvironment(1);
    34.  
    35.         //TODO 2 读取kafka播放日志数据
    36.         String topicName = "dwd_traffic_play_pre_process";
    37.         String groupId = "dwd_learn_play";
    38.         DataStreamSource playSource = env.fromSource(KafkaUtil.getKafkaConsumer(topicName, groupId), WatermarkStrategy.noWatermarks(), "learn_play");
    39.  
    40.         //TODO 3 清洗转换
    41.         SingleOutputStreamOperator learnBeanStream = playSource.flatMap(new FlatMapFunction() {
    42.             @Override
    43.             public void flatMap(String value, Collector out) throws Exception {
    44.                 try {
    45.                     JSONObject jsonObject = JSON.parseObject(value);
    46.                     JSONObject common = jsonObject.getJSONObject("common");
    47.                     JSONObject appVideo = jsonObject.getJSONObject("appVideo");
    48.                     Long ts = jsonObject.getLong("ts");
    49.                     DwdLearnPlayBean learnPlayBean = DwdLearnPlayBean.builder()
    50.                             .provinceId(common.getString("ar"))
    51.                             .brand(common.getString("ba"))
    52.                             .channel(common.getString("ch"))
    53.                             .isNew(common.getString("is_new"))
    54.                             .model(common.getString("md"))
    55.                             .machineId(common.getString("mid"))
    56.                             .operatingSystem(common.getString("os"))
    57.                             .sourceId(common.getString("sc"))
    58.                             .sessionId(common.getString("sid"))
    59.                             .userId(common.getString("uid"))
    60.                             .versionCode(common.getString("vc"))
    61.                             .playSec(appVideo.getInteger("play_sec"))
    62.                             .videoId(appVideo.getString("video_id"))
    63.                             .positionSec(appVideo.getInteger("position_sec"))
    64.                             .ts(ts)
    65.                             .build();
    66.                     out.collect(learnPlayBean);
    67.                 } catch (Exception e) {
    68.                     e.printStackTrace();
    69.                 }
    70.             }
    71.         });
    72.  
    73.         //TODO 4 添加水位线
    74.         SingleOutputStreamOperator withWatermarkStream = learnBeanStream.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(
    75.                 new SerializableTimestampAssigner() {
    76.                     @Override
    77.                     public long extractTimestamp(DwdLearnPlayBean element, long recordTimestamp) {
    78.                         return element.getTs();
    79.                     }
    80.                 }
    81.         ));
    82.  
    83.         //TODO 5 按照会话id分组
    84.         KeyedStream keyedStream = withWatermarkStream.keyBy(new KeySelector() {
    85.             @Override
    86.             public String getKey(DwdLearnPlayBean value) throws Exception {
    87.                 return value.getSessionId();
    88.             }
    89.         });
    90.  
    91.         //TODO 6 聚合统计
    92.         WindowedStream windowStream = keyedStream.window(EventTimeSessionWindows.withGap(Time.seconds(3L)));
    93.         SingleOutputStreamOperator reducedStream = windowStream.reduce(
    94.                 new ReduceFunction() {
    95.                     @Override
    96.                     public DwdLearnPlayBean reduce(DwdLearnPlayBean value1, DwdLearnPlayBean value2) throws Exception {
    97.                         value1.setPlaySec(value1.getPlaySec() + value2.getPlaySec());
    98.                         if (value2.getTs() > value1.getTs()) {
    99.                             value1.setPositionSec(value2.getPositionSec());
    100.                         }
    101.                         return value1;
    102.                     }
    103.                 }, new ProcessWindowFunction() {
    104.                     @Override
    105.                     public void process(String key, Context context, Iterable elements, Collector out) throws Exception {
    106.                         for (DwdLearnPlayBean element : elements) {
    107.                             out.collect(element);
    108.                         }
    109.                     }
    110.                 }
    111.         );
    112.  
    113.         //TODO 7 转换结构
    114.         SingleOutputStreamOperator jsonStrStream = reducedStream.map(JSON::toJSONString);
    115.  
    116.         //TODO 8 输出到kafka主题Kafka dwd_learn_play
    117.         String targetTopic = "dwd_learn_play";
    118.         jsonStrStream.sinkTo(KafkaUtil.getKafkaProducer(targetTopic,"learn_pay_trans"));
    119.  
    120.         //TODO 9 执行任务
    121.         env.execute();
    122.     }
    123. }

    P048

    先启动消费者DwdLearnPlay，再mock数据。

    kafka没有消费到数据，DwdLearnPlay：将并发改为1(TODO 1)、改时间(TODO 6，时间改为3s)，窗口和并发调小一些。

    同一个人看的同一个视频，时间不一样，看的位置也不一样。

    [atguigu@node001 ~]$ kafka-console-consumer.sh --bootstrap-server node001:9092 --topic dwd_learn_play
    1. [atguigu@node001 ~]$ cd /opt/module/data_mocker/01-onlineEducation/
    2. [atguigu@node001 01-onlineEducation]$ ll
    3. 总用量 30460
    4. -rw-rw-r-- 1 atguigu atguigu     2223 9月  19 10:43 application.yml
    5. -rw-rw-r-- 1 atguigu atguigu  4057995 7月  25 10:28 edu0222.sql
    6. -rw-rw-r-- 1 atguigu atguigu 27112074 7月  25 10:28 edu2021-mock-2022-06-18.jar
    7. drwxrwxr-x 2 atguigu atguigu     4096 11月  2 11:13 log
    8. -rw-rw-r-- 1 atguigu atguigu     1156 7月  25 10:44 logback.xml
    9. -rw-rw-r-- 1 atguigu atguigu      633 7月  25 10:45 path.json
    10. [atguigu@node001 01-onlineEducation]$ java -jar edu2021-mock-2022-06-18.jar 
    11. SLF4J: Class path contains multiple SLF4J bindings.
    12. SLF4J: Found binding in [jar:file:/opt/module/data_mocker/01-onlineEducation/edu2021-mock-2022-06-18.jar!/BOOT-INF/lib/logback-classic-1.2.3.jar!/org/slf4j/impl/StaticLoggerBinder.class]
    13. SLF4J: Found binding in [jar:file:/opt/module/data_mocker/01-onlineEducation/edu2021-mock-2022-06-18.jar!/BOOT-INF/lib/slf4j-log4j12-1.7.7.jar!/org/slf4j/impl/StaticLoggerBinder.class]
    1. {"brand":"Xiaomi","channel":"xiaomi","isNew":"0","machineId":"mid_293","model":"Xiaomi Mix2 ","operatingSystem":"Android 10.0","playSec":30,"positionSec":690,"provinceId":"18","sessionId":"a1fb6d22-f8ef-40e6-89c2-262cd5a351be","sourceId":"1","ts":1645460612085,"userId":"46","versionCode":"v2.1.134","videoId":"108"}
    2. {"brand":"Xiaomi","channel":"xiaomi","isNew":"0","machineId":"mid_293","model":"Xiaomi Mix2 ","operatingSystem":"Android 10.0","playSec":30,"positionSec":720,"provinceId":"18","sessionId":"a1fb6d22-f8ef-40e6-89c2-262cd5a351be","sourceId":"1","ts":1645460642085,"userId":"46","versionCode":"v2.1.134","videoId":"108"}
    3. {
    4.     "brand":"Xiaomi",
    5.     "channel":"xiaomi",
    6.     "isNew":"0",
    7.     "machineId":"mid_293",
    8.     "model":"Xiaomi Mix2 ",
    9.     "operatingSystem":"Android 10.0",
    10.     "playSec":30,
    11.     "positionSec":690,
    12.     "provinceId":"18",
    13.     "sessionId":"a1fb6d22-f8ef-40e6-89c2-262cd5a351be",
    14.     "sourceId":"1",
    15.     "ts":1645460612085,
    16.     "userId":"46",
    17.     "versionCode":"v2.1.134",
    18.     "videoId":"108"
    19. }

    P049

    9.5 用户域用户登录事务事实表

    9.5.1 主要任务

    读取页面日志数据，筛选用户登录记录，写入 Kafka 用户登录主题。

    9.5.2 思路分析

    9.5.3 图解

    P050

    DwdUserUserLogin

    //TODO 1 创建环境设置状态后端

    //TODO 2 读取kafka的dwd_traffic_page_log主题数据

    //TODO 3 过滤及转换

    //TODO 4 添加水位线

    //TODO 5 按照会话id分组

    P051

    DwdUserUserLogin、DwdUserUserLoginBean

    1. package com.atguigu.edu.realtime.app.dwd.log;
    2.  
    3. import com.alibaba.fastjson.JSON;
    4. import com.alibaba.fastjson.JSONObject;
    5. import com.atguigu.edu.realtime.bean.DwdUserUserLoginBean;
    6. import com.atguigu.edu.realtime.util.DateFormatUtil;
    7. import com.atguigu.edu.realtime.util.EnvUtil;
    8. import com.atguigu.edu.realtime.util.KafkaUtil;
    9. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
    10. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    11. import org.apache.flink.api.common.functions.FlatMapFunction;
    12. import org.apache.flink.api.common.functions.MapFunction;
    13. import org.apache.flink.api.common.state.StateTtlConfig;
    14. import org.apache.flink.api.common.state.ValueState;
    15. import org.apache.flink.api.common.state.ValueStateDescriptor;
    16. import org.apache.flink.api.common.time.Time;
    17. import org.apache.flink.api.java.functions.KeySelector;
    18. import org.apache.flink.configuration.Configuration;
    19. import org.apache.flink.streaming.api.datastream.DataStreamSource;
    20. import org.apache.flink.streaming.api.datastream.KeyedStream;
    21. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    22. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    23. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    24. import org.apache.flink.util.Collector;
    25.  
    26. import java.time.Duration;
    27.  
    28. /**
    29.  * @author yhm
    30.  * @create 2023-04-23 16:02
    31.  */
    32. public class DwdUserUserLogin {
    33.     public static void main(String[] args) throws Exception {
    34.         //TODO 1 创建环境设置状态后端
    35.         StreamExecutionEnvironment env = EnvUtil.getExecutionEnvironment(1);
    36.  
    37.         //TODO 2 读取kafka的dwd_traffic_page_log主题数据
    38.         String topicName = "dwd_traffic_page_log";
    39.         String groupId = "dwd_user_user_login";
    40.         DataStreamSource pageStream = env.fromSource(KafkaUtil.getKafkaConsumer(topicName, groupId), WatermarkStrategy.noWatermarks(), "user_login");
    41.  
    42.         //TODO 3 过滤及转换
    43.         SingleOutputStreamOperator jsonObjStream = pageStream.flatMap(new FlatMapFunction() {
    44.             @Override
    45.             public void flatMap(String value, Collector out) throws Exception {
    46.                 try {
    47.                     JSONObject jsonObject = JSON.parseObject(value);
    48.                     if (jsonObject.getJSONObject("common").getString("uid") != null) {
    49.                         out.collect(jsonObject);
    50.                     }
    51.                 } catch (Exception e) {
    52.                     e.printStackTrace();
    53.                 }
    54.             }
    55.         });
    56.  
    57.         //TODO 4 添加水位线
    58.         SingleOutputStreamOperator withWaterMarkStream = jsonObjStream.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5L)).withTimestampAssigner(new SerializableTimestampAssigner() {
    59.             @Override
    60.             public long extractTimestamp(JSONObject element, long recordTimestamp) {
    61.                 return element.getLong("ts");
    62.             }
    63.         }));
    64.  
    65.         //TODO 5 按照会话id分组
    66.         KeyedStream keyedStream = withWaterMarkStream.keyBy(new KeySelector() {
    67.             @Override
    68.             public String getKey(JSONObject value) throws Exception {
    69.                 return value.getJSONObject("common").getString("mid");
    70.             }
    71.         });
    72.  
    73.         //TODO 6 使用状态找出每个会话第一条数据
    74.         SingleOutputStreamOperator firstStream = keyedStream.process(new KeyedProcessFunction() {
    75.             ValueState firstLoginDtState;
    76.  
    77.             @Override
    78.             public void open(Configuration parameters) throws Exception {
    79.                 super.open(parameters);
    80.                 ValueStateDescriptor valueStateDescriptor = new ValueStateDescriptor<>("first_login_dt", JSONObject.class);
    81.                 // 添加状态存活时间
    82.                 valueStateDescriptor.enableTimeToLive(StateTtlConfig
    83.                         .newBuilder(Time.days(1L))
    84.                         .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    85.                         .build());
    86.                 firstLoginDtState = getRuntimeContext().getState(valueStateDescriptor);
    87.             }
    88.  
    89.             @Override
    90.             public void processElement(JSONObject jsonObject, Context ctx, Collector out) throws Exception {
    91.                 // 处理数据
    92.                 // 获取状态
    93.                 JSONObject firstLoginDt = firstLoginDtState.value();
    94.                 Long ts = jsonObject.getLong("ts");
    95.                 if (firstLoginDt == null) {
    96.                     firstLoginDtState.update(jsonObject);
    97.                     // 第一条数据到的时候开启定时器
    98.                     ctx.timerService().registerEventTimeTimer(ts + 10 * 1000L);
    99.                 } else {
    100.                     Long lastTs = firstLoginDt.getLong("ts");
    101.                     if (ts < lastTs) {
    102.                         firstLoginDtState.update(jsonObject);
    103.                     }
    104.                 }
    105.             }
    106.  
    107.             @Override
    108.             public void onTimer(long timestamp, OnTimerContext ctx, Collector out) throws Exception {
    109.                 super.onTimer(timestamp, ctx, out);
    110.                 out.collect(firstLoginDtState.value());
    111.             }
    112.         });
    113.  
    114.         //TODO 7 转换结构
    115.         SingleOutputStreamOperator mapStream = firstStream.map(new MapFunction() {
    116.             @Override
    117.             public String map(JSONObject jsonObj) throws Exception {
    118.                 JSONObject common = jsonObj.getJSONObject("common");
    119.                 Long ts = jsonObj.getLong("ts");
    120.                 String loginTime = DateFormatUtil.toYmdHms(ts);
    121.                 String dateId = loginTime.substring(0, 10);
    122.  
    123.                 DwdUserUserLoginBean dwdUserUserLoginBean = DwdUserUserLoginBean.builder()
    124.                         .userId(common.getString("uid"))
    125.                         .dateId(dateId)
    126.                         .loginTime(loginTime)
    127.                         .channel(common.getString("ch"))
    128.                         .provinceId(common.getString("ar"))
    129.                         .versionCode(common.getString("vc"))
    130.                         .midId(common.getString("mid"))
    131.                         .brand(common.getString("ba"))
    132.                         .model(common.getString("md"))
    133.                         .sourceId(common.getString("sc"))
    134.                         .operatingSystem(common.getString("os"))
    135.                         .ts(ts)
    136.                         .build();
    137.                 return JSON.toJSONString(dwdUserUserLoginBean);
    138.             }
    139.         });
    140.  
    141.         //TODO 8 输出数据
    142.         String sinkTopic = "dwd_user_user_login";
    143.         mapStream.sinkTo(KafkaUtil.getKafkaProducer(sinkTopic, "user_login_trans"));
    144.  
    145.         //TODO 9 执行任务
    146.         env.execute();
    147.     }
    148. }

    P052

    [atguigu@node001 ~]$ kafka-console-consumer.sh --bootstrap-server node001:9092 --topic dwd_user_user_login
    1. [atguigu@node001 ~]$ cd /opt/module/data_mocker/01-onlineEducation/
    2. [atguigu@node001 01-onlineEducation]$ java -jar edu2021-mock-2022-06-18.jar
  • 相关阅读:
    E. Qpwoeirut and Vertices（思维 + MST + lca + 线段树）
    国庆作业2
    行为模式－责任链模式
    day25-登录和代理IP
    基于windows、GDAL2.2.3版本和Java集成安装和使用GDAL库的方法
    dc_shell的change_names命令，信号[]/_
    video元素与audio元素详解
    带你一分钟看懂 “Docker”
    Windows 系统下怎么获取 UDP 本机地址
    linux网桥简单理解和持久化配置
  • 原文地址：https://blog.csdn.net/weixin_44949135/article/details/134178909