[大数据]Flink入门第八课：DataStream api的watermark的相关操作

package com.atguigu.Ctime;

import com.atguigu.Fbeans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
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.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;


/**
 * watermark是一条带有时间戳的单调递增的数据记录，它出现在数据流中意味着watermark之前的数据已全部到齐，
 *      此时如果没有设置allowedLateness，直接结束窗口触发最终计算；如果设置了，则先触发一次计算，
 *      然后按allowedLateness规定时间进行等待，等待过程中数据来一条计算一次，等待结束立马关闭窗口。
 *      窗口关闭后watermark以广播的方式传递会下游算子。由于任务是并行的，下游算子接收到watermark可能有多个，
 *      在当前subtask以最少小的为准逐个处理watermark。
 * 设置watermark一般在分组之前，allowedLateness是在分组之后。
 *
 * watermaker=当前最大事件时间-允许延迟时间。
 * watermaker>=窗口结束时间的时候，该窗口会触发计算。
 *
 * datastaream.assignTimestampsAndWatermaker(new T())
 *  T可以是AscendingTimestampExtractor(数据升序时用，不需要指定延迟时间)
 *  也可以是BoundedOutOfOrdernessTimestampExtractor(数据有界乱序时用)。
 *  在T的方法内获取实体类的时间字段作为事件时间，方法内时间单位是毫秒。
 *
 *  watermark可以来一条数据生成一次，也可以按时间间隔来生成。
 *  默认200ms生成一次可以通过env.getConfig().setAutoWatermarkInterval()来设置。
 *
 * 只能解决一定程度上的数据乱序和数据延时问题，完全解决需要使用OutputTag+sideOutputLateData侧道输出机制，
 * 相当于将allowedLateness配置的最大延迟之后的数据收集放入OutputTag中再输出。当然也可以对其做一些计算再输出。
 *
 * 设置事件时间：env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 数据流中必须有事件时间字段
 *
 */
public class AWaterMarkTest {
    public static void main(String[] args) throws Exception {
        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取数据并包装成pojo
        DataStreamSource<String> inputStream = env. socketTextStream("localhost",7777);
        DataStream<SensorReading> mapStream = inputStream.map(line -> {
            String[] splits = line.split(",");
            //这儿toString是为了数据传输时方便使用simpleStringSchema
            return new SensorReading(new String(splits[0]), new Long(splits[1]), new Double(splits[2]));
        });

        /**
         * 分配watermark
         *  watermark支持两个类：BoundedOutOfOrdernessTimestampExtractor和AscendingTimestampExtractor
         *  两个类都是用来设置watermark的，watermark的时间单位是毫秒。但是：
         *      前者处理有界乱序数据，需要设置好延迟时间。
         *      后者处理有序数据，不需要设置延迟时间。一般前者用得更多。
         *  watermark可以来一条数据生成一次（用得少），也可以按间隔来生成，默认是200ms生成一次。
         *  我们可以通过env.getConfig().setAutoWatermarkInterval(Time.seconds(200))来手动指定生成间隔。
         *
         */
        SingleOutputStreamOperator<SensorReading> waterStream = mapStream.assignTimestampsAndWatermarks(
                //watermark的延迟是10s
                new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(10)) {
                    @Override
                    public long extractTimestamp(SensorReading sen) {
                        //提取字段中的事件时间。watermark时间单位是毫秒
                        return sen.getTimestamp() * 1000L;
                    }
                });
        /*mapStream.assignTimestampsAndWatermarks(
                new AscendingTimestampExtractor<SensorReading>() {
            @Override
            public long extractAscendingTimestamp(SensorReading sen) {

                return sen.getTimestamp()*1000L;
            }
        })*/

        /**
         * 开始计算,获取最近十五秒的最小温度记录,不包含watermark允许最大延迟1min
         *  超过最大延迟的数据加入侧输出流后再输出
         *
         */
        OutputTag<SensorReading> tag= new OutputTag<SensorReading>("late"); //封装延时数据
        SingleOutputStreamOperator<SensorReading> result = waterStream.keyBy(SensorReading::getId)
                .timeWindow(Time.seconds(15))
                .allowedLateness(Time.minutes(1)) //watermark到达窗口结束时间后还允许的最大延迟
                .sideOutputLateData(tag)  //侧道输出机制
                .minBy(2);

        //输出并执行
        result.print("测试watermark操作");
        result.getSideOutput(tag).print("late");//打印侧输出流数据，这儿也可以对侧输出流做一些计算再输出
        env.execute("测试watermark操作");
    }
}