目录

案例三 会话窗口

需求

代码实现

案例三 会话窗口

需求

设置会话超时时间为10s,10s内没有数据到来,则触发上个窗口的计算

​​​​​​​代码实现

      package cn.itcast.window;
      import lombok.AllArgsConstructor;
      import lombok.Data;
      import lombok.NoArgsConstructor;
      import org.apache.flink.api.common.functions.MapFunction;
      import org.apache.flink.streaming.api.datastream.DataStreamSource;
      import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
      import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
      import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
      import org.apache.flink.streaming.api.windowing.time.Time;
      /**
       * Author itcast
       * Desc
       * nc -lk 9999
       * 有如下数据表示:
       * 信号灯编号和通过该信号灯的车的数量
      9,3
      9,2
      9,7
      4,9
      2,6
      1,5
      2,3
      5,7
      5,4
       * 需求:设置会话超时时间为10s,10s内没有数据到来,则触发上个窗口的计算(前提是上一个窗口得有数据!)
       */
      public class WindowDemo03_SessionWindow {
          public static void main(String[] args) throws Exception {
              //1.env
              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
              //2.Source
              DataStreamSource<String> socketDS = env.socketTextStream("node1", 9999);
              //3.Transformation
              //将9,3转为CartInfo(9,3)
              SingleOutputStreamOperator<CartInfo> cartInfoDS = socketDS.map(new MapFunction<String, CartInfo>() {
                  @Override
                  public CartInfo map(String value) throws Exception {
                      String[] arr = value.split(",");
                      return new CartInfo(arr[0], Integer.parseInt(arr[1]));
                  }
              });
              //需求:设置会话超时时间为10s,10s内没有数据到来,则触发上个窗口的计算(前提是上一个窗口得有数据!)
              SingleOutputStreamOperator<CartInfo> result = cartInfoDS.keyBy(CartInfo::getSensorId)
                      .window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)))
                      .sum("count");
              //4.Sink
              result.print();
              //5.execute
              env.execute();
          }
          @Data
          @AllArgsConstructor
          @NoArgsConstructor
          public static class CartInfo {
              private String sensorId;//信号灯id
              private Integer count;//通过该信号灯的车的数量
          }
      }
  
 

文章来源: lansonli.blog.csdn.net，作者：Lansonli，版权归原作者所有，如需转载，请联系作者。

原文链接：lansonli.blog.csdn.net/article/details/116279229