Flink：解决数据延迟产生的问题-Watermark

偏执的太偏执、 2024-03-23 12:00 78阅读 0赞

### **Watermark案例** ###

**步骤**：

1、获取数据源

2、转化

3、声明水印（watermark）

4、分组聚合，调用window的操作

5、保存处理结果

数据源：

01,1586489566000

01,1586489567000

01,1586489568000

01,1586489569000

01,1586489570000

01,1586489571000

01,1586489572000

01,1586489573000

即为

2020-04-10 11:32:46

2020-04-10 11:32:47

2020-04-10 11:32:48

2020-04-10 11:32:49

2020-04-10 11:32:50

import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    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.functions.AssignerWithPeriodicWatermarks;
    import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
    import org.apache.flink.streaming.api.watermark.Watermark;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    import org.apache.flink.util.Collector;
    
    import javax.annotation.Nullable;
    import java.text.SimpleDateFormat;
    import java.util.Iterator;
    
    public class WaterMarkDemo {
    
        public static void main(String[] args) throws Exception {
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // 并行度为1
            env.setParallelism(1);
    
            // 设定flink程序处理数据的时候用那种时间数据
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            DataStreamSource<String> data = env.socketTextStream("", 7777);
            // MapFunction第一个泛型表示输入的数据类型；第二个泛型表示输出的数据类型（可以自定义）
            SingleOutputStreamOperator<Tuple2<String, Long>> maped = data.map(new MapFunction<String, Tuple2<String, Long>>() {
    
                public Tuple2<String, Long> map(String value) throws Exception {
                    String[] split = value.split(",");
                    return new Tuple2<String, Long>(split[0], Long.valueOf(split[1]));
                }
            });
    
            // 做出水印watermark
            // 将数据源中的每一个数据增加一个水印时间数据
            // AssignerWithPeriodicWatermarks对传入的数据周期性的打印水印
            SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = maped.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {
                
                // 当前时间戳数据
                Long currentMaxStamp = 0L;
                // 允许的最大延迟时间，此处为2秒
                Long maxOrderOut = 2000L;
    
                @Nullable
                public Watermark getCurrentWatermark() {
                    return new Watermark(currentMaxStamp - maxOrderOut);
                }
    
                /**
                 * 生成时间戳 ----- EventTime
                 * @param element
                 * @param l
                 * @return
                 */
                public long extractTimestamp(Tuple2<String, Long> element, long l) {
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                    Long f1 = element.f1;
                    currentMaxStamp = Math.max(f1, currentMaxStamp);
                    System.out.println("EventTime:" + f1 + "...f1.format:" + sdf.format(f1) + "...watermark:" + getCurrentWatermark().getTimestamp() + "...watermark.format:" + sdf.format(getCurrentWatermark().getTimestamp()));
                    return f1;
                }
            });
    
            // 根据二元组第一个元素分组，创建时间滚动窗口，窗口时间大小为2秒
            // WindowFunction第一个泛型就是输入数据的数据类型；第二个泛型就是输出数据的数据类型；
            // 第三个就是key的数据类型；第四个就是窗口的数据类型
            SingleOutputStreamOperator<String> result = watermarks.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(2))).apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
                public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<Tuple2<String, Long>> iterable, Collector<String> collector) throws Exception {
    
                    // 窗口开始时间
                    long start = timeWindow.getStart();
                    // 窗口结束时间
                    long end = timeWindow.getEnd();
                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                    String startTime = simpleDateFormat.format(start);
                    String endTime = simpleDateFormat.format(end);
    
                    Iterator<Tuple2<String, Long>> iterator = iterable.iterator();
                    ArrayList<Long> list = new ArrayList<Long>();
    
                    while (iterator.hasNext()) {
                        Tuple2<String, Long> next = iterator.next();
                        String key = next.f0;
                        Long f1 = next.f1;
                        list.add(f1);
                    }
    
                    Collections.sort(list);
                    // 第一条数据的时间
                    String firstTime = simpleDateFormat.format(list.get(0));
                    // 最后一条数据的时间
                    String lastTime = simpleDateFormat.format(list.get(list.size() - 1));
    
                    collector.collect("startTime:" + startTime + "...endTime:" + endTime + "...firstTime:" + firstTime + "...lastTime:" + lastTime);
                }
            });
    
            result.print();
            env.execute();
        }
    }

**注意：**

当使用EventTimeWindow时，所有的Window在EventTime的时间轴上进行划分，

也就是说，在Window启动后，会根据初始的EventTime时间每隔一段时间划分一个窗口，

如果Window大小是3秒，那么1分钟内会把Window划分为如下的形式：

\[00:00:00,00:00:03)

\[00:00:03,00:00:06)

\[00:00:03,00:00:09)

\[00:00:03,00:00:12)

\[00:00:03,00:00:15)

\[00:00:03,00:00:18)

\[00:00:03,00:00:21)

\[00:00:03,00:00:24)

...

\[00:00:57,00:00:42)

\[00:00:57,00:00:45)

\[00:00:57,00:00:48)

...

如果Window大小是10秒，则Window会被分为如下的形式：

\[00:00:00,00:00:10)

\[00:00:10,00:00:20)

...

\[00:00:50,00:01:00)

**说明：**

1.窗口是左闭右开的，形式为：\[window\_start\_time,window\_end\_time)。

2.Window的设定基于第一条消息的事件时间，也就是说，Window会一直按照指定的时间间隔进行划分，不论这个Window中有没有数据，EventTime在这个Window期间的数据会进入这个Window。

3.Window会不断产生，属于这个Window范围的数据会被不断加入到Window中，所有未被触发的Window都会等待触发，只要Window还没触发，属于这个Window范围的数据就会一直被加入到Window中，直到Window被触发才会停止数据的追加，而当Window触发之后才接受到的属于被触发Window的数据会被丢弃。

4.Window会在以下的条件满足时被触发执行：

(1)在\[window\_start\_time,window\_end\_time)窗口中有数据存在

(2)watermark时间 >= window\_end\_time；

5.一般会设置水印时间，比事件时间小几秒钟,表示最大允许数据延迟达到多久

(即水印时间 = 事件时间 - 允许延迟时间)

当接收到的 水印时间 >= 窗口结束时间且窗口内有数据，则触发计算