第一章.基于埋点日志的网络流量统计

1.数据准备

UserBehavior.csv

2.指定时间范围内网站总浏览量(PV)的统计

需求分析:实现一个网站总浏览量的统计,可以使用滚动时间窗口来实现,实时统计每小时内的网站PV

  1. package com.atguigu.flink.day07;
  3. import com.atguigu.flink.day07.pojo.UserBehavior;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.api.common.functions.AggregateFunction;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
  8. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  9. import org.apache.flink.streaming.api.windowing.time.Time;
  10. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  11. import org.apache.flink.util.Collector;
  13. import java.time.Duration;
  14. import java.util.Date;
  16. /**
  17.  * 需求:指定时间范围内网站总浏览量(PV)的统计
  18.  */
  20. public class $01_PVCount {
  21.     public static void main(String[] args) throws Exception {
  22.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  23.         env.readTextFile("input/UserBehavior.csv")
  24.                 .map(line -> {
  25.                     String[] data = line.split(",");
  26.                     return new UserBehavior(
  27.                             Long.parseLong(data[0]),
  28.                             Long.parseLong(data[1]),
  29.                             Integer.parseInt(data[2]),
  30.                             data[3], Long.parseLong(data[4] ) * 1000
  31.                     );
  32.                 })
  33.                 .assignTimestampsAndWatermarks(
  34.                         WatermarkStrategy
  35.                                .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  36.                                 .withTimestampAssigner((ub,ts)-> ub.getTimestamp())
  37.                 )
  38.                 .filter(ub -> "pv".equals(ub.getBehavior()))
  39.                 .keyBy(ub -> ub.getBehavior())
  40.                 .window(TumblingEventTimeWindows.of(Time.minutes(30)))
  41.                 .aggregate(
  42.                         new AggregateFunction<UserBehavior, Long, Long>() {
  43.                             @Override
  44.                             public Long createAccumulator() {
  45.                                 return 0L;
  46.                             }
  48.                             @Override
  49.                             public Long add(UserBehavior value, Long accumulator) {
  50.                                 return accumulator + 1;
  51.                             }
  53.                             @Override
  54.                             public Long getResult(Long accumulator) {
  55.                                 return accumulator;
  56.                             }
  58.                             @Override
  59.                             public Long merge(Long a, Long b) {
  60.                                 return null;
  61.                             }
  62.                         },
  63.                         new WindowFunction<Long, String, String, TimeWindow>() {
  64.                             @Override
  65.                             public void apply(String key, TimeWindow window, Iterable<Long> input, Collector<String> out) throws Exception {
  66.                                 Long pv = input.iterator().next();
  67.                                 String msg = new Date(window.getStart()) + " " + new Date(window.getEnd()) + " " + pv;
  68.                                 out.collect(msg);
  69.                             }
  70.                         }
  71.                 )
  72.                 .print();
  73.         env.execute();
  75.     }
  76. }

day07[Flink流处理高阶编程实战] - 图1

3.指定时间范围内网站独立访客数(UV)的统计

  1. package com.atguigu.flink.day07;
  3. import com.atguigu.flink.day07.pojo.UserBehavior;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.api.common.functions.AggregateFunction;
  6. import org.apache.flink.api.common.state.MapState;
  7. import org.apache.flink.api.common.state.MapStateDescriptor;
  8. import org.apache.flink.api.common.typeinfo.Types;
  9. import org.apache.flink.configuration.Configuration;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  12. import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
  13. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  14. import org.apache.flink.streaming.api.windowing.time.Time;
  15. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  16. import org.apache.flink.util.Collector;
  18. import java.time.Duration;
  19. import java.util.ArrayList;
  20. import java.util.Arrays;
  21. import java.util.Date;
  22. import java.util.Iterator;
  24. /**
  25.  * 需求:指定时间范围内网站独立访客数(UV)的统计
  26.  */
  28. public class $02_UVCount {
  29.     public static void main(String[] args) throws Exception {
  30.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  31.         env.readTextFile("input/UserBehavior.csv")
  32.                 .map(line -> {
  33.                     String[] data = line.split(",");
  34.                     return new UserBehavior(
  35.                             Long.parseLong(data[0]),
  36.                             Long.parseLong(data[1]),
  37.                             Integer.parseInt(data[2]),
  38.                             data[3], Long.parseLong(data[4] ) * 1000
  39.                     );
  40.                 })
  41.                 .assignTimestampsAndWatermarks(
  42.                         WatermarkStrategy
  43.                                .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  44.                                 .withTimestampAssigner((ub,ts)-> ub.getTimestamp())
  45.                 )
  46.                 .filter(ub -> "pv".equals(ub.getBehavior()))
  47.                 .keyBy(ub -> ub.getBehavior())
  48.                 .window(TumblingEventTimeWindows.of(Time.minutes(30)))
  49.                 .process(new ProcessWindowFunction<UserBehavior, String, String, TimeWindow>() {
  51.                     private MapState<Long, String> userIdState;
  53.                     @Override
  54.                     public void open(Configuration parameters) throws Exception {
  55.                         userIdState = getRuntimeContext().getMapState(new MapStateDescriptor<Long, String>(
  56.                                 "userIdState",
  57.                                 Types.LONG,
  58.                                 Types.STRING
  59.                         ));
  60.                     }
  62.                     @Override
  63.                     public void process(String s, Context context, Iterable<UserBehavior> elements, Collector<String> out) throws Exception {
  64.                         userIdState.clear();
  65.                         for (UserBehavior element : elements) {
  66.                             //所有用户的id存入到状态中,自动去重
  67.                             userIdState.put(element.getUserId(),"a");
  69.                         }
  70.                         Iterator<Long> iterator = userIdState.keys().iterator();
  71.                         ArrayList<Long> arrayList = new ArrayList<>();
  72.                         while(iterator.hasNext()){
  73.                             arrayList.add(iterator.next());
  74.                         }
  76.                         out.collect(context.window() + " " + arrayList.size());
  77.                     }
  78.                 })
  79.                 .print();
  80.         env.execute();
  85.     }
  86. }

day07[Flink流处理高阶编程实战] - 图2

第二章.电商数据分析

  1.     电商平台的用户行为频繁且比较复杂,系统上线运行一段时间后,可以收集到大量的用户行为数据,进而利用大数据技术进行深入挖掘和分析,得到感兴趣的商业指标并增强对风险的控制
  2.     电商用户行为数据多样,整体可以分为用户行为数据(日志数据)和业务行为数据两大类
  3.     用户的行为习惯数据包括了用户的登录方式,上线的时间点及时长,点击和浏览页面,页面停留时间以及页面跳转等等,我们可以从中进行流量统计和热门商品的统计,也可以深入挖掘用户的特征,这些数据往往可以从web服务器日志中直接读取到
  4.     二业务行为数据就是用户在电商平台中针对每个业务(通常是某个具体商品)所做的具体操作,我们一般会在业务系统中相应的位置埋点,然后收集日志进行分析

1.数据准备

UserBehavior.csv

2.实时热门商品统计

需求分析: 每隔30分钟输出最近一小时内点击量最多的前N个商品

  • 最近一小时:窗口长度
  • 每隔5分钟:窗口滑动步长
  • 时间:使用event-time(事件时间)

这里准备一个javabean用来保存商品信息

  1. package com.atguigu.flink.day07.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. @Data
  8. @AllArgsConstructor
  9. @NoArgsConstructor
  10. public class HotItem {
  11.     private Long itemId;
  12.     private Long windowEndTime;
  13.     private Long count;   
  14. }

具体实现代码

  1. package com.atguigu.flink.day07;
  3. import com.atguigu.flink.day07.pojo.HotItem;
  4. import com.atguigu.flink.day07.pojo.UserBehavior;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.AggregateFunction;
  7. import org.apache.flink.api.common.state.ListState;
  8. import org.apache.flink.api.common.state.ListStateDescriptor;
  9. import org.apache.flink.api.common.state.MapState;
  10. import org.apache.flink.api.common.state.MapStateDescriptor;
  11. import org.apache.flink.api.common.typeinfo.Types;
  12. import org.apache.flink.configuration.Configuration;
  13. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  14. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  15. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  16. import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
  17. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  18. import org.apache.flink.streaming.api.windowing.time.Time;
  19. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  20. import org.apache.flink.util.Collector;
  22. import java.time.Duration;
  23. import java.util.ArrayList;
  24. import java.util.Iterator;
  26. /**
  27.  * 需求:每隔30分钟输出最近一小时内点击量最多的前N个商品
  28.  */
  30. public class $03_TopN {
  31.     public static void main(String[] args) throws Exception {
  32.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  33.         env.readTextFile("input/UserBehavior.csv")
  34.                 .map(line -> {
  35.                     String[] data = line.split(",");
  36.                     return new UserBehavior(
  37.                             Long.parseLong(data[0]),
  38.                             Long.parseLong(data[1]),
  39.                             Integer.parseInt(data[2]),
  40.                             data[3], Long.parseLong(data[4] ) * 1000
  41.                     );
  42.                 })
  43.                 .assignTimestampsAndWatermarks(
  44.                         WatermarkStrategy
  45.                                .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  46.                                 .withTimestampAssigner((ub,ts)-> ub.getTimestamp())
  47.                 )
  48.                 .filter(ub -> "pv".equals(ub.getBehavior()))
  49.                 .keyBy(ub -> ub.getItemId())
  50.                 .window(SlidingEventTimeWindows.of(Time.hours(1),Time.minutes(30)))
  51.                 .aggregate(
  52.                         new AggregateFunction<UserBehavior, Long, Long>() {
  53.                             @Override
  54.                             public Long createAccumulator() {
  55.                                 return 0L;
  56.                             }
  58.                             @Override
  59.                             public Long add(UserBehavior value, Long accumulator) {
  60.                                 return accumulator + 1;
  61.                             }
  63.                             @Override
  64.                             public Long getResult(Long accumulator) {
  65.                                 return accumulator;
  66.                             }
  68.                             @Override
  69.                             public Long merge(Long a, Long b) {
  70.                                 return null;
  71.                             }
  72.                         },
  73.                         new ProcessWindowFunction<Long, HotItem, Long, TimeWindow>() {
  74.                             @Override
  75.                             public void process(Long aLong, Context context, Iterable<Long> elements, Collector<HotItem> out) throws Exception {
  76.                                 out.collect(new HotItem(aLong,context.window().getEnd(),elements.iterator().next()));
  77.                             }
  78.                         }
  81.                 )
  82.                 .keyBy(hotItem  -> hotItem.getWindowEndTime())
  83.                 .process(new KeyedProcessFunction<Long, HotItem, String>() {
  86.                     private ListState<HotItem> hotItemState;
  88.                     @Override
  89.                     public void open(Configuration parameters) throws Exception {
  90.                         hotItemState = getRuntimeContext().getListState(new ListStateDescriptor<HotItem>(
  91.                                 "hotItemState",
  92.                                 HotItem.class
  93.                         ));
  94.                     }
  96.                     @Override
  97.                     public void processElement(HotItem value, Context ctx, Collector<String> out) throws Exception {
  98.                         //每来一条数据,把数据存入到状态,等到定时器触发的时候,进行排序
  99.                         if(!hotItemState.get().iterator().hasNext()){
  100.                             //现在是来了第一条数据:注册定时器
  101.                             ctx.timerService().registerEventTimeTimer(value.getWindowEndTime() + 5000L);
  102.                         }
  103.                         hotItemState.add(value);
  104.                     }
  105.                     @Override
  106.                     public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
  107.                         Iterator<HotItem> iterator = hotItemState.get().iterator();
  108.                         ArrayList<HotItem> hotItems = new ArrayList<>();
  109.                         while(iterator.hasNext()){
  110.                             hotItems.add(iterator.next());
  111.                         }
  112.                         hotItems.sort((o1,o2)->o2.getCount().compareTo(o1.getCount()));
  113.                         StringBuilder stringBuilder = new StringBuilder();
  114.                         stringBuilder.append("-------------------");
  115.                         for (int i = 0,count = Math.min(3, hotItems.size());i<count;i++){
  116.                             stringBuilder.append(hotItems.get(i)).append("\t");
  117.                         }
  118.                         out.collect(stringBuilder.toString());
  120.                     }
  121.                 })
  122.                 .print();
  123.         env.execute();
  131.     }
  132. }

day07[Flink流处理高阶编程实战] - 图3

第三章.页面广告分析

1.数据准备

AdClickLog.csv

2.黑名单过滤

  1.     我们进行的点击量计算,同一用户的重复点击是会叠加计算的
  2.     在实际场景中,同一用户确实可能反复点开同一个广告,这也说明了用户对广告更大的兴趣,但是如果用户在一段时间频繁的点击广告,这显然不是一个正常行为,有刷点击量的嫌疑
  3.     所以我们可以对一段时间内(比如一天内)的用户点击行为进行约束,如果对同一个广告点击超过一定限额(比如100次)应该把该用户加入黑名单并报警,此后其点击行为不应该再统计

两个功能

  • 告警:使用侧输出流
  • 已经进入黑名单的用户的广告点击记录不再进行统计
  1. package com.atguigu.flink.day07;
  3. import com.atguigu.flink.day07.pojo.AdsClickLog;
  4. import com.atguigu.flink.day07.pojo.HotItem;
  5. import com.atguigu.flink.day07.pojo.UserBehavior;
  6. import org.apache.commons.lang.math.LongRange;
  7. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  8. import org.apache.flink.api.common.functions.AggregateFunction;
  9. import org.apache.flink.api.common.functions.ReduceFunction;
  10. import org.apache.flink.api.common.state.*;
  11. import org.apache.flink.api.common.typeinfo.Types;
  12. import org.apache.flink.configuration.Configuration;
  13. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  14. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  15. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  16. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  17. import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
  18. import org.apache.flink.streaming.api.windowing.time.Time;
  19. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  20. import org.apache.flink.util.Collector;
  21. import org.apache.flink.util.OutputTag;
  23. import java.text.SimpleDateFormat;
  24. import java.time.Duration;
  25. import java.util.ArrayList;
  26. import java.util.Iterator;
  28. /**
  29.  * 需求:黑名单过滤(时间间隔为一天,广告点击次数为100次)
  30.  */
  32. public class $04_BlackList {
  33.     public static void main(String[] args) throws Exception {
  34.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  35.         env.setParallelism(1);
  36.         SingleOutputStreamOperator<String> normal = env.readTextFile("input/AdClickLog.csv")
  37.                 .map(line -> {
  38.                     String[] data = line.split(",");
  39.                     return new AdsClickLog(
  40.                             Long.parseLong(data[0]),
  41.                             Long.parseLong(data[1]),
  42.                             data[2],
  43.                             data[3],
  44.                             Long.parseLong(data[4])
  46.                     );
  47.                 })
  48.                 .assignTimestampsAndWatermarks(
  49.                         WatermarkStrategy
  50.                                 .<AdsClickLog>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  51.                                 .withTimestampAssigner((log, ts) -> log.getTimestamp())
  52.                 )
  53.                 .keyBy(log -> log.getUserId() + ":" + log.getAdsId())
  54.                 .process(new KeyedProcessFunction<String, AdsClickLog, String>() {
  55.                     private ValueState<Boolean> warnState;
  56.                     private ReducingState<Long> countState;
  57.                     //定义一个状态存储年月日,如果新的数据和状态一致则是同一天,否则就是第二天
  58.                     private ValueState<String> yesterdayState;
  60.                     @Override
  61.                     public void open(Configuration parameters) throws Exception {
  62.                         countState = getRuntimeContext().getReducingState(
  63.                                 new ReducingStateDescriptor<Long>(
  64.                                         "countState",
  65.                                         new ReduceFunction<Long>() {
  66.                                             @Override
  67.                                             public Long reduce(Long value1, Long value2) throws Exception {
  68.                                                 return value1 + value2;
  69.                                             }
  70.                                         },
  71.                                         Types.LONG
  73.                                 )
  74.                         );
  75.                         warnState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("warnState", Types.BOOLEAN));
  76.                         yesterdayState = getRuntimeContext().getState(new ValueStateDescriptor<String>("yesterdayState", Types.STRING));
  77.                     }
  79.                     @Override
  80.                     public void processElement(AdsClickLog value, Context ctx, Collector<String> out) throws Exception {
  82.                         String today = new SimpleDateFormat("yyyyMMdd").format(value.getTimestamp());
  83.                         String yesterday = yesterdayState.value();
  84.                         //如果到了第二天,清空状态点击量和黑名单重新计算
  85.                         if (!today.equals(yesterday)) {
  86.                             countState.clear();
  87.                             warnState.clear();
  88.                             yesterdayState.update(today);
  89.                         }
  90.                         if (warnState.value() == null) {
  91.                             //说明没有预警,需要对用户的操作进行统计
  92.                             countState.add(1L);
  93.                         }
  94.                         String msg = value.getUserId() + "对广告:" + value.getAdsId() + "的点击量:" + countState.get();
  95.                         if (countState.get() > 99) {
  96.                             if (warnState.value() == null) {
  97.                                 ctx.output(new OutputTag<String>("black") {
  98.                                 }, msg + ",超过了阈值100,加入黑名单");
  99.                                 warnState.update(true);
  100.                             }
  101.                         } else {
  102.                             out.collect(msg);
  103.                         }
  104.                     }
  105.                 });
  106.         normal.print("正常");
  107.         normal.getSideOutput(new OutputTag<String>("black"){}).print("black");
  108.         env.execute();
  111.     }
  112. }

day07[Flink流处理高阶编程实战] - 图4

第四章.恶意登录监控

  1.     对于网站而言,用户登录并不是频繁的业务操作,如果一个用户短时间内频繁登录失败,就有可能是出现了程序的恶意攻击,比如密码暴力破解
  2.     因此,我们考虑,应该对用户的登录失败动作进行统计,具体来说,如果一个同一用户在2s之内连续两次登录失败,就认为存在恶意登录的风险,输出相关的信息进行报警提示,这是电商网站,也是几乎所有网站风控的基本一环

1.数据准备

LoginLog.csv

2.封装数据的JavaBean

  1. package com.atguigu.flink.day07.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. @Data
  8. @NoArgsConstructor
  9. @AllArgsConstructor
  10. public class LoginEvent {
  11.     private Long userId;
  12.     private String ip;
  13.     private String eventType;
  14.     private Long eventTime;
  15. }

3.具体实现代码

  1. package com.atguigu.flink.day07;
  3. import com.atguigu.flink.day07.pojo.LoginEvent;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  6. import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
  7. import org.apache.flink.util.Collector;
  9. import java.util.ArrayList;
  10. import java.util.Iterator;
  12. /**
  13.  * 需求:恶意登录监控(同一用户在2s之内连续两次登录失败)
  14.  */
  16. public class $05_Login {
  17.     public static void main(String[] args) throws Exception {
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.         env.readTextFile("input/LoginLog.csv")
  21.                 .map(line -> {
  22.                     String[] data = line.split(",");
  23.                     return new LoginEvent(
  24.                             Long.parseLong(data[0]),
  25.                             data[1],
  26.                             data[2],
  27.                             Long.parseLong(data[3] )* 1000L
  28.                     );
  29.                 })
  30.                 .keyBy(log->log.getUserId())
  31.                 .countWindow(2,1)
  32.                 .process(new ProcessWindowFunction<LoginEvent, String, Long, GlobalWindow>() {
  33.                     @Override
  34.                     public void process(Long aLong, Context context, Iterable<LoginEvent> elements, Collector<String> out) throws Exception {
  35.                         Iterator<LoginEvent> iterator = elements.iterator();
  36.                         ArrayList<LoginEvent> loginEvents = new ArrayList<>();
  37.                         while(iterator.hasNext()){
  38.                             loginEvents.add(iterator.next());
  39.                         }
  40.                         if(loginEvents.size() ==2){
  41.                             //如果长度为2才需要进行计算
  42.                             LoginEvent event1 = loginEvents.get(0);
  43.                             LoginEvent event2 = loginEvents.get(1);
  44.                             String eventType1 = event1.getEventType();
  45.                             String eventType2 = event2.getEventType();
  46.                             Long time1 = event1.getEventTime();
  47.                             Long time2 = event2.getEventTime();
  48.                             if("fail".equals(eventType1)
  49.                             && "fail".equals(eventType2)
  50.                             && Math.abs(time1 -time2)<=2000){
  51.                                 out.collect(aLong + "在恶意登录,请注意....");
  52.                             }
  53.                         }
  54.                     }
  55.                 })
  56.                 .print();
  57.         env.execute();
  59.     }
  60. }

day07[Flink流处理高阶编程实战] - 图5

day07[Flink流处理高阶编程实战] - 图6

第五章.订单支付实时监控

  1.     在电商网站中,订单的支付作为直接与营销收入挂钩的一环,在业务流程中非常重要
  2.     对于订单而言,为了正确控制业务过程,也为了增加用户的支付意愿,网站一般会设置一个支付失效时间,超过一段时间不支付的订单就会被取消
  3.     另外,对于订单的支付,我们还应保证用户支付的正确性,这可以通过第三方支付平台的交易数据来做一个实时对账

1.数据准备

OrderLog.csv

2.封装数据的JavaBean

  1. package com.atguigu.flink.day07.pojo;
  2. import lombok.AllArgsConstructor;
  3. import lombok.Data;
  4. import lombok.NoArgsConstructor;
  5. @Data
  6. @AllArgsConstructor
  7. @NoArgsConstructor
  8. public class OrderEvent {
  9.     private Long orderId;
  10.     private String eventType;
  11.     private String txId;
  12.     private Long eventTime;
  13. }

3.具体代码实现

  1. package com.atguigu.flink.day07;
  4. import com.atguigu.flink.day07.pojo.OrderEvent;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.state.ValueState;
  7. import org.apache.flink.api.common.state.ValueStateDescriptor;
  9. import org.apache.flink.configuration.Configuration;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  12. import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
  13. import org.apache.flink.streaming.api.windowing.time.Time;
  15. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  16. import org.apache.flink.util.Collector;
  18. import java.time.Duration;
  19. import java.util.ArrayList;
  20. import java.util.Iterator;
  22. /**
  23.  * 需求:订单支付实时监控
  24.  */
  26. public class $06_Order {
  27.     public static void main(String[] args) throws Exception {
  28.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  29.         env.setParallelism(1);
  30.         env.readTextFile("input/OrderLog.csv")
  31.                 .map(line -> {
  32.                     String[] data = line.split(",");
  33.                     return new OrderEvent(
  34.                             Long.valueOf(data[0]),
  35.                             data[1],
  36.                             data[2],
  37.                             Long.parseLong(data[3]) * 1000
  38.                     );
  39.                 })
  40.                 .assignTimestampsAndWatermarks(
  41.                         WatermarkStrategy
  42.                                 .<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  43.                                 .withTimestampAssigner((event, ts) -> event.getEventTime())
  44.                 )
  45.                 .keyBy(orderEvent -> orderEvent.getOrderId())
  46.                 .window(EventTimeSessionWindows.withGap(Time.minutes(45)))
  47.                 .process(new ProcessWindowFunction<OrderEvent, String, Long, TimeWindow>() {
  49.                     private ValueState<OrderEvent> creatState;
  51.                     @Override
  52.                     public void open(Configuration parameters) throws Exception {
  53.                         creatState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>(
  54.                                 "creatState",
  55.                                 OrderEvent.class
  56.                         ));
  57.                     }
  59.                     @Override
  60.                     public void process(Long key, Context context, Iterable<OrderEvent> elements, Collector<String> out) throws Exception {
  61.                         /**
  62.                          * 如果订单支付和创建都正常情况:则集合中有两个元素
  63.                          *  如果只有一个create,则表示没有key,或者pay的时间超过了45分钟
  64.                          */
  65.                         Iterator<OrderEvent> iterator = elements.iterator();
  66.                         ArrayList<OrderEvent> orderEvents = new ArrayList<>();
  67.                         while (iterator.hasNext()) {
  68.                             orderEvents.add(iterator.next());
  69.                         }
  70.                         if (orderEvents.size() == 2) {
  71.                             out.collect("订单:" + key + "正常");
  72.                         } else {
  73.                             //如果长度不是2,则一定是1
  74.                             OrderEvent event = orderEvents.get(0);
  75.                             if ("create".equals(event.getEventType())) {
  76.                                 //这个session窗口只有create,没有pay,要么真的没有,要么超时
  77.                                 creatState.update(event);
  78.                             } else {
  79.                                 if (creatState.value() != null) {
  80.                                     //表示create来过但是pay晚于create超过了45分钟,他们没有在一个窗口中
  81.                                     out.collect("订单:" + key + "超时支付,请检查系统漏洞");
  82.                                 } else {
  83.                                     out.collect("订单:" + key + " 只有pay没有create, 请检测系统漏洞");
  84.                                 }
  85.                             }
  86.                         }
  87.                     }
  88.                 })
  89.                 .print();
  90.         env.execute();
  93.     }
  94. }

day07[Flink流处理高阶编程实战] - 图7