flink的watermark参考配置

需求描述：每隔5秒，计算最近10秒单词出现的次数。
TimeWindow实现

/*** 每隔5秒计算最近10秒单词出现的次数*/
public class TimeWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> dataStream = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<Tuple2<String, Integer>> result = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {String[] fields = line.split(",");for (String word : fields) {out.collect(new Tuple2<>(word, 1));}}}).keyBy(0).timeWindow(Time.seconds(10), Time.seconds(5)).sum(1);result.print().setParallelism(1);env.execute("TimeWindowWordCount");}
}

ProcessWindowFunction

/*** 每隔5秒计算最近10秒单词出现的次数*/
public class TimeWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> dataStream = env.socketTextStream("10.148.15.10", 8888);SingleOutputStreamOperator<Tuple2<String, Integer>> result = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {String[] fields = line.split(",");for (String word : fields) {out.collect(new Tuple2<>(word, 1));}}}).keyBy(0).timeWindow(Time.seconds(10), Time.seconds(5)).process(new SumProcessWindowFunction());result.print().setParallelism(1);env.execute("TimeWindowWordCount");}/*** IN, OUT, KEY, W* IN：输入的数据类型* OUT：输出的数据类型* Key：key的数据类型（在Flink里面，String用Tuple表示）* W：Window的数据类型*/public static class SumProcessWindowFunction extendsProcessWindowFunction<Tuple2<String,Integer>,Tuple2<String,Integer>,Tuple,TimeWindow> {FastDateFormat dataFormat = FastDateFormat.getInstance("HH:mm:ss");/*** 当一个window触发计算的时候会调用这个方法* @param tuple key* @param context operator的上下文* @param elements 指定window的所有元素* @param out 用户输出*/@Overridepublic void process(Tuple tuple, Context context, Iterable<Tuple2<String, Integer>> elements,Collector<Tuple2<String, Integer>> out) {System.out.println("当天系统的时间："+dataFormat.format(System.currentTimeMillis()));System.out.println("Window的处理时间："+dataFormat.format(context.currentProcessingTime()));System.out.println("Window的开始时间："+dataFormat.format(context.window().getStart()));System.out.println("Window的结束时间："+dataFormat.format(context.window().getEnd()));int sum = 0;for (Tuple2<String, Integer> ele : elements) {sum += 1;}// 输出单词出现的次数out.collect(Tuple2.of(tuple.getField(0), sum));}}
}

Time的种类

针对stream数据中的时间，可以分为以下三种：
Event Time：事件产生的时间，它通常由事件中的时间戳描述。
Ingestion time：事件进入Flink的时间
Processing Time：事件被处理时当前系统的时间

Process Time Window

需求：每隔5秒计算最近10秒的单词出现的次数

自定义source，模拟：第 13 秒的时候连续发送 2 个事件，第 16 秒的时候再发送 1 个事件

/*** 每隔5秒计算最近10秒单词出现的次数*/
public class TimeWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<String> dataStream = env.addSource(new TestSouce());SingleOutputStreamOperator<Tuple2<String, Integer>> result = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {String[] fields = line.split(",");for (String word : fields) {out.collect(new Tuple2<>(word, 1));}}}).keyBy(0).timeWindow(Time.seconds(10), Time.seconds(5)).process(new SumProcessWindowFunction());result.print().setParallelism(1);env.execute("TimeWindowWordCount");}public static class TestSouce implements SourceFunction<String>{FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");@Overridepublic void run(SourceContext<String> ctx) throws Exception {// 控制大约在 10 秒的倍数的时间点发送事件String currTime = String.valueOf(System.currentTimeMillis());while (Integer.valueOf(currTime.substring(currTime.length() - 4)) > 100) {currTime = String.valueOf(System.currentTimeMillis());continue;}System.out.println("开始发送事件的时间：" + dateFormat.format(System.currentTimeMillis()));// 第 13 秒发送两个事件TimeUnit.SECONDS.sleep(13);ctx.collect("hadoop," + System.currentTimeMillis());// 产生了一个事件，但是由于网络原因，事件没有发送ctx.collect("hadoop," + System.currentTimeMillis());// 第 16 秒发送一个事件TimeUnit.SECONDS.sleep(3);ctx.collect("hadoop," + System.currentTimeMillis());TimeUnit.SECONDS.sleep(300);}@Overridepublic void cancel() {}}/*** IN, OUT, KEY, W* IN：输入的数据类型* OUT：输出的数据类型* Key：key的数据类型（在Flink里面，String用Tuple表示）* W：Window的数据类型*/public static class SumProcessWindowFunction extendsProcessWindowFunction<Tuple2<String,Integer>,Tuple2<String,Integer>,Tuple,TimeWindow> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");/*** 当一个window触发计算的时候会调用这个方法* @param tuple key* @param context operator的上下文* @param elements 指定window的所有元素* @param out 用户输出*/@Overridepublic void process(Tuple tuple, Context context, Iterable<Tuple2<String, Integer>> elements,Collector<Tuple2<String, Integer>> out) {//            System.out.println("当天系统的时间："+dateFormat.format(System.currentTimeMillis()));
//
//            System.out.println("Window的处理时间："+dateFormat.format(context.currentProcessingTime()));
//            System.out.println("Window的开始时间："+dateFormat.format(context.window().getStart()));
//            System.out.println("Window的结束时间："+dateFormat.format(context.window().getEnd()));int sum = 0;for (Tuple2<String, Integer> ele : elements) {sum += 1;}// 输出单词出现的次数out.collect(Tuple2.of(tuple.getField(0), sum));}}
}

Process Time Window

自定义source，模拟：第 13 秒的时候连续发送 2 个事件，但是有一个事件确实在第13秒的发送出去了，另外一个事件因为某种原因在19秒的时候才发送出去，第 16 秒的时候再发送 1 个事件

/*** 每隔5秒计算最近10秒单词出现的次数*/
public class TimeWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<String> dataStream = env.addSource(new TestSouce());SingleOutputStreamOperator<Tuple2<String, Integer>> result = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {String[] fields = line.split(",");for (String word : fields) {out.collect(new Tuple2<>(word, 1));}}}).keyBy(0).timeWindow(Time.seconds(10), Time.seconds(5)).process(new SumProcessWindowFunction());result.print().setParallelism(1);env.execute("TimeWindowWordCount");}/*** 模拟：第 13 秒的时候连续发送 2 个事件，第 16 秒的时候再发送 1 个事件*/public static class TestSouce implements SourceFunction<String>{FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");@Overridepublic void run(SourceContext<String> ctx) throws Exception {// 控制大约在 10 秒的倍数的时间点发送事件String currTime = String.valueOf(System.currentTimeMillis());while (Integer.valueOf(currTime.substring(currTime.length() - 4)) > 100) {currTime = String.valueOf(System.currentTimeMillis());continue;}System.out.println("开始发送事件的时间：" + dateFormat.format(System.currentTimeMillis()));// 第 13 秒发送两个事件TimeUnit.SECONDS.sleep(13);ctx.collect("hadoop," + System.currentTimeMillis());// 产生了一个事件，但是由于网络原因，事件没有发送String event = "hadoop," + System.currentTimeMillis();// 第 16 秒发送一个事件TimeUnit.SECONDS.sleep(3);ctx.collect("hadoop," + System.currentTimeMillis());// 第 19 秒的时候发送TimeUnit.SECONDS.sleep(3);ctx.collect(event);TimeUnit.SECONDS.sleep(300);}@Overridepublic void cancel() {}}/*** IN, OUT, KEY, W* IN：输入的数据类型* OUT：输出的数据类型* Key：key的数据类型（在Flink里面，String用Tuple表示）* W：Window的数据类型*/public static class SumProcessWindowFunction extendsProcessWindowFunction<Tuple2<String,Integer>,Tuple2<String,Integer>,Tuple,TimeWindow> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");/*** 当一个window触发计算的时候会调用这个方法* @param tuple key* @param context operator的上下文* @param elements 指定window的所有元素* @param out 用户输出*/@Overridepublic void process(Tuple tuple, Context context, Iterable<Tuple2<String, Integer>> elements,Collector<Tuple2<String, Integer>> out) {//            System.out.println("当天系统的时间："+dateFormat.format(System.currentTimeMillis()));
//
//            System.out.println("Window的处理时间："+dateFormat.format(context.currentProcessingTime()));
//            System.out.println("Window的开始时间："+dateFormat.format(context.window().getStart()));
//            System.out.println("Window的结束时间："+dateFormat.format(context.window().getEnd()));int sum = 0;for (Tuple2<String, Integer> ele : elements) {sum += 1;}// 输出单词出现的次数out.collect(Tuple2.of(tuple.getField(0), sum));}}
}

使用Event Time处理无序

使用Event Time处理

/*** 每隔5秒计算最近10秒单词出现的次数*/
public class TimeWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//步骤一：设置时间类型env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);DataStreamSource<String> dataStream = env.addSource(new TestSouce());dataStream.map(new MapFunction<String, Tuple2<String,Long>>() {@Overridepublic Tuple2<String, Long> map(String line) throws Exception {String[] fields = line.split(",");return new Tuple2<>(fields[0],Long.valueOf(fields[1]));}//步骤二：获取数据里面的event Time}).assignTimestampsAndWatermarks(new EventTimeExtractor() ).keyBy(0).timeWindow(Time.seconds(10), Time.seconds(5)).process(new SumProcessWindowFunction()).print().setParallelism(1);env.execute("TimeWindowWordCount");}public static class TestSouce implements SourceFunction<String>{FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");@Overridepublic void run(SourceContext<String> ctx) throws Exception {// 控制大约在 10 秒的倍数的时间点发送事件String currTime = String.valueOf(System.currentTimeMillis());while (Integer.valueOf(currTime.substring(currTime.length() - 4)) > 100) {currTime = String.valueOf(System.currentTimeMillis());continue;}System.out.println("开始发送事件的时间：" + dateFormat.format(System.currentTimeMillis()));// 第 13 秒发送两个事件TimeUnit.SECONDS.sleep(13);ctx.collect("hadoop," + System.currentTimeMillis());// 产生了一个事件，但是由于网络原因，事件没有发送String event = "hadoop," + System.currentTimeMillis();// 第 16 秒发送一个事件TimeUnit.SECONDS.sleep(3);ctx.collect("hadoop," + System.currentTimeMillis());// 第 19 秒的时候发送TimeUnit.SECONDS.sleep(3);ctx.collect(event);TimeUnit.SECONDS.sleep(300);}@Overridepublic void cancel() {}}/*** IN, OUT, KEY, W* IN：输入的数据类型* OUT：输出的数据类型* Key：key的数据类型（在Flink里面，String用Tuple表示）* W：Window的数据类型*/public static class SumProcessWindowFunction extendsProcessWindowFunction<Tuple2<String,Long>,Tuple2<String,Integer>,Tuple,TimeWindow> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");/*** 当一个window触发计算的时候会调用这个方法* @param tuple key* @param context operator的上下文* @param elements 指定window的所有元素* @param out 用户输出*/@Overridepublic void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,Collector<Tuple2<String, Integer>> out) {//            System.out.println("当天系统的时间："+dateFormat.format(System.currentTimeMillis()));
//
//            System.out.println("Window的处理时间："+dateFormat.format(context.currentProcessingTime()));
//            System.out.println("Window的开始时间："+dateFormat.format(context.window().getStart()));
//            System.out.println("Window的结束时间："+dateFormat.format(context.window().getEnd()));int sum = 0;for (Tuple2<String, Long> ele : elements) {sum += 1;}// 输出单词出现的次数out.collect(Tuple2.of(tuple.getField(0), sum));}}private static class EventTimeExtractorimplements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");// 拿到每一个事件的 Event Time@Overridepublic long extractTimestamp(Tuple2<String, Long> element,long previousElementTimestamp) {return element.f1;}@Nullable@Overridepublic Watermark getCurrentWatermark() {return new Watermark(System.currentTimeMillis());}}
}

现在我们第三个window的结果已经计算准确了，但是我们还是没有彻底的解决问题。接下来就需要我们使用WaterMark机制来解决了。

使用WaterMark机制解决无序

/*** 每隔5秒计算最近10秒单词出现的次数*/
public class TimeWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//步骤一：设置时间类型env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);DataStreamSource<String> dataStream = env.addSource(new TestSouce());dataStream.map(new MapFunction<String, Tuple2<String,Long>>() {@Overridepublic Tuple2<String, Long> map(String line) throws Exception {String[] fields = line.split(",");return new Tuple2<>(fields[0],Long.valueOf(fields[1]));}//步骤二：获取数据里面的event Time}).assignTimestampsAndWatermarks(new EventTimeExtractor() ).keyBy(0).timeWindow(Time.seconds(10), Time.seconds(5)).process(new SumProcessWindowFunction()).print().setParallelism(1);env.execute("TimeWindowWordCount");}public static class TestSouce implements SourceFunction<String>{FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");@Overridepublic void run(SourceContext<String> ctx) throws Exception {// 控制大约在 10 秒的倍数的时间点发送事件String currTime = String.valueOf(System.currentTimeMillis());while (Integer.valueOf(currTime.substring(currTime.length() - 4)) > 100) {currTime = String.valueOf(System.currentTimeMillis());continue;}System.out.println("开始发送事件的时间：" + dateFormat.format(System.currentTimeMillis()));// 第 13 秒发送两个事件TimeUnit.SECONDS.sleep(13);ctx.collect("hadoop," + System.currentTimeMillis());// 产生了一个事件，但是由于网络原因，事件没有发送String event = "hadoop," + System.currentTimeMillis();// 第 16 秒发送一个事件TimeUnit.SECONDS.sleep(3);ctx.collect("hadoop," + System.currentTimeMillis());// 第 19 秒的时候发送TimeUnit.SECONDS.sleep(3);ctx.collect(event);TimeUnit.SECONDS.sleep(300);}@Overridepublic void cancel() {}}/*** IN, OUT, KEY, W* IN：输入的数据类型* OUT：输出的数据类型* Key：key的数据类型（在Flink里面，String用Tuple表示）* W：Window的数据类型*/public static class SumProcessWindowFunction extendsProcessWindowFunction<Tuple2<String,Long>,Tuple2<String,Integer>,Tuple,TimeWindow> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");/*** 当一个window触发计算的时候会调用这个方法* @param tuple key* @param context operator的上下文* @param elements 指定window的所有元素* @param out 用户输出*/@Overridepublic void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,Collector<Tuple2<String, Integer>> out) {//            System.out.println("当天系统的时间："+dateFormat.format(System.currentTimeMillis()));
//
//            System.out.println("Window的处理时间："+dateFormat.format(context.currentProcessingTime()));
//            System.out.println("Window的开始时间："+dateFormat.format(context.window().getStart()));
//            System.out.println("Window的结束时间："+dateFormat.format(context.window().getEnd()));int sum = 0;for (Tuple2<String, Long> ele : elements) {sum += 1;}// 输出单词出现的次数out.collect(Tuple2.of(tuple.getField(0), sum));}}private static class EventTimeExtractorimplements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");// 拿到每一个事件的 Event Time@Overridepublic long extractTimestamp(Tuple2<String, Long> element,long previousElementTimestamp) {return element.f1;}@Nullable@Overridepublic Watermark getCurrentWatermark() {//window延迟5秒触发return new Watermark(System.currentTimeMillis() - 5000);}}
}

WaterMark的周期

/*** 每隔5秒计算最近10秒单词出现的次数*/
public class TimeWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//步骤一：设置时间类型env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//设置waterMark产生的周期为1senv.getConfig().setAutoWatermarkInterval(1000);DataStreamSource<String> dataStream = env.addSource(new TestSouce());dataStream.map(new MapFunction<String, Tuple2<String,Long>>() {@Overridepublic Tuple2<String, Long> map(String line) throws Exception {String[] fields = line.split(",");return new Tuple2<>(fields[0],Long.valueOf(fields[1]));}//步骤二：获取数据里面的event Time}).assignTimestampsAndWatermarks(new EventTimeExtractor() ).keyBy(0).timeWindow(Time.seconds(10), Time.seconds(5)).process(new SumProcessWindowFunction()).print().setParallelism(1);env.execute("TimeWindowWordCount");}public static class TestSouce implements SourceFunction<String>{FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");@Overridepublic void run(SourceContext<String> ctx) throws Exception {// 控制大约在 10 秒的倍数的时间点发送事件String currTime = String.valueOf(System.currentTimeMillis());while (Integer.valueOf(currTime.substring(currTime.length() - 4)) > 100) {currTime = String.valueOf(System.currentTimeMillis());continue;}System.out.println("开始发送事件的时间：" + dateFormat.format(System.currentTimeMillis()));// 第 13 秒发送两个事件TimeUnit.SECONDS.sleep(13);ctx.collect("hadoop," + System.currentTimeMillis());// 产生了一个事件，但是由于网络原因，事件没有发送String event = "hadoop," + System.currentTimeMillis();// 第 16 秒发送一个事件TimeUnit.SECONDS.sleep(3);ctx.collect("hadoop," + System.currentTimeMillis());// 第 19 秒的时候发送TimeUnit.SECONDS.sleep(3);ctx.collect(event);TimeUnit.SECONDS.sleep(300);}@Overridepublic void cancel() {}}/*** IN, OUT, KEY, W* IN：输入的数据类型* OUT：输出的数据类型* Key：key的数据类型（在Flink里面，String用Tuple表示）* W：Window的数据类型*/public static class SumProcessWindowFunction extendsProcessWindowFunction<Tuple2<String,Long>,Tuple2<String,Integer>,Tuple,TimeWindow> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");/*** 当一个window触发计算的时候会调用这个方法* @param tuple key* @param context operator的上下文* @param elements 指定window的所有元素* @param out 用户输出*/@Overridepublic void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,Collector<Tuple2<String, Integer>> out) {int sum = 0;for (Tuple2<String, Long> ele : elements) {sum += 1;}// 输出单词出现的次数out.collect(Tuple2.of(tuple.getField(0), sum));}}private static class EventTimeExtractorimplements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");// 拿到每一个事件的 Event Time@Overridepublic long extractTimestamp(Tuple2<String, Long> element,long previousElementTimestamp) {//这个方法是每获取到一个数据就会被调用一次。return element.f1;}@Nullable@Overridepublic Watermark getCurrentWatermark() {/*** WasterMark会周期性的产生，默认就是每隔200毫秒产生一个**         设置 watermark 产生的周期为 1000ms*         env.getConfig().setAutoWatermarkInterval(1000);*///window延迟5秒触发System.out.println("water mark...");return new Watermark(System.currentTimeMillis() - 5000);}}
}

得到并打印每隔 3 秒钟统计前 3 秒内的相同的 key 的所有的事件

/*** 得到并打印每隔 3 秒钟统计前 3 秒内的相同的 key 的所有的事件*/
public class WaterMarkWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//步骤一：设置时间类型env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//设置waterMark产生的周期为1senv.getConfig().setAutoWatermarkInterval(1000);DataStreamSource<String> dataStream = env.socketTextStream("10.148.15.10", 8888);dataStream.map(new MapFunction<String, Tuple2<String,Long>>() {@Overridepublic Tuple2<String, Long> map(String line) throws Exception {String[] fields = line.split(",");return new Tuple2<>(fields[0],Long.valueOf(fields[1]));}//步骤二：获取数据里面的event Time}).assignTimestampsAndWatermarks(new EventTimeExtractor() ).keyBy(0).timeWindow(Time.seconds(3)).process(new SumProcessWindowFunction()).print().setParallelism(1);env.execute("TimeWindowWordCount");}/*** IN, OUT, KEY, W* IN：输入的数据类型* OUT：输出的数据类型* Key：key的数据类型（在Flink里面，String用Tuple表示）* W：Window的数据类型*/public static class SumProcessWindowFunction extendsProcessWindowFunction<Tuple2<String,Long>,String,Tuple,TimeWindow> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");/*** 当一个window触发计算的时候会调用这个方法* @param tuple key* @param context operator的上下文* @param elements 指定window的所有元素* @param out 用户输出*/@Overridepublic void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,Collector<String> out) {System.out.println("处理时间：" + dateFormat.format(context.currentProcessingTime()));System.out.println("window start time : " + dateFormat.format(context.window().getStart()));List<String> list = new ArrayList<>();for (Tuple2<String, Long> ele : elements) {list.add(ele.toString() + "|" + dateFormat.format(ele.f1));}out.collect(list.toString());System.out.println("window end time  : " + dateFormat.format(context.window().getEnd()));}}private static class EventTimeExtractorimplements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");private long currentMaxEventTime = 0L;private long maxOutOfOrderness = 10000; // 最大允许的乱序时间 10 秒// 拿到每一个事件的 Event Time@Overridepublic long extractTimestamp(Tuple2<String, Long> element,long previousElementTimestamp) {long currentElementEventTime = element.f1;currentMaxEventTime = Math.max(currentMaxEventTime, currentElementEventTime);System.out.println("event = " + element+ "|" + dateFormat.format(element.f1) // Event Time+ "|" + dateFormat.format(currentMaxEventTime)  // Max Event Time+ "|" + dateFormat.format(getCurrentWatermark().getTimestamp())); // Current Watermarkreturn currentElementEventTime;}@Nullable@Overridepublic Watermark getCurrentWatermark() {/*** WasterMark会周期性的产生，默认就是每隔200毫秒产生一个**         设置 watermark 产生的周期为 1000ms*         env.getConfig().setAutoWatermarkInterval(1000);*///window延迟5秒触发System.out.println("water mark...");return new Watermark(currentMaxEventTime - maxOutOfOrderness);}}
}
收集迟到的数据
/*** 得到并打印每隔 3 秒钟统计前 3 秒内的相同的 key 的所有的事件* 收集迟到太多的数据*/
public class WaterMarkWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//步骤一：设置时间类型env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//设置waterMark产生的周期为1senv.getConfig().setAutoWatermarkInterval(1000);// 保存迟到的，会被丢弃的数据OutputTag<Tuple2<String, Long>> outputTag =new OutputTag<Tuple2<String, Long>>("late-date"){};DataStreamSource<String> dataStream = env.socketTextStream("10.148.15.10", 8888);SingleOutputStreamOperator<String> result = dataStream.map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> map(String line) throws Exception {String[] fields = line.split(",");return new Tuple2<>(fields[0], Long.valueOf(fields[1]));}//步骤二：获取数据里面的event Time}).assignTimestampsAndWatermarks(new EventTimeExtractor()).keyBy(0).timeWindow(Time.seconds(3))// .allowedLateness(Time.seconds(2)) // 允许事件迟到 2 秒.sideOutputLateData(outputTag) // 保存迟到太多的数据.process(new SumProcessWindowFunction());//打印正常的数据result.print();//获取迟到太多的数据DataStream<String> lateDataStream= result.getSideOutput(outputTag).map(new MapFunction<Tuple2<String, Long>, String>() {@Overridepublic String map(Tuple2<String, Long> stringLongTuple2) throws Exception {return "迟到的数据：" + stringLongTuple2.toString();}});lateDataStream.print();env.execute("TimeWindowWordCount");}/*** IN, OUT, KEY, W* IN：输入的数据类型* OUT：输出的数据类型* Key：key的数据类型（在Flink里面，String用Tuple表示）* W：Window的数据类型*/public static class SumProcessWindowFunction extendsProcessWindowFunction<Tuple2<String,Long>,String,Tuple,TimeWindow> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");/*** 当一个window触发计算的时候会调用这个方法* @param tuple key* @param context operator的上下文* @param elements 指定window的所有元素* @param out 用户输出*/@Overridepublic void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,Collector<String> out) {System.out.println("处理时间：" + dateFormat.format(context.currentProcessingTime()));System.out.println("window start time : " + dateFormat.format(context.window().getStart()));List<String> list = new ArrayList<>();for (Tuple2<String, Long> ele : elements) {list.add(ele.toString() + "|" + dateFormat.format(ele.f1));}out.collect(list.toString());System.out.println("window end time  : " + dateFormat.format(context.window().getEnd()));}}private static class EventTimeExtractorimplements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");private long currentMaxEventTime = 0L;private long maxOutOfOrderness = 10000; // 最大允许的乱序时间 10 秒// 拿到每一个事件的 Event Time@Overridepublic long extractTimestamp(Tuple2<String, Long> element,long previousElementTimestamp) {long currentElementEventTime = element.f1;currentMaxEventTime = Math.max(currentMaxEventTime, currentElementEventTime);System.out.println("event = " + element+ "|" + dateFormat.format(element.f1) // Event Time+ "|" + dateFormat.format(currentMaxEventTime)  // Max Event Time+ "|" + dateFormat.format(getCurrentWatermark().getTimestamp())); // Current Watermarkreturn currentElementEventTime;}@Nullable@Overridepublic Watermark getCurrentWatermark() {/*** WasterMark会周期性的产生，默认就是每隔200毫秒产生一个**         设置 watermark 产生的周期为 1000ms*         env.getConfig().setAutoWatermarkInterval(1000);*/System.out.println("water mark...");return new Watermark(currentMaxEventTime - maxOutOfOrderness);}}
}

多并行度下的WaterMark
一个window可能会接受到多个waterMark，我们以最小的为准。

/**
* 得到并打印每隔 3 秒钟统计前 3 秒内的相同的 key 的所有的事件
* 测试多并行度
*/
public class WaterMarkWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//把并行度设置为2env.setParallelism(2);//步骤一：设置时间类型env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//设置waterMark产生的周期为1senv.getConfig().setAutoWatermarkInterval(1000);// 保存迟到的，会被丢弃的数据OutputTag<Tuple2<String, Long>> outputTag =new OutputTag<Tuple2<String, Long>>("late-date"){};DataStreamSource<String> dataStream = env.socketTextStream("10.148.15.10", 8888);SingleOutputStreamOperator<String> result = dataStream.map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> map(String line) throws Exception {String[] fields = line.split(",");return new Tuple2<>(fields[0], Long.valueOf(fields[1]));}//步骤二：获取数据里面的event Time}).assignTimestampsAndWatermarks(new EventTimeExtractor()).keyBy(0).timeWindow(Time.seconds(3))// .allowedLateness(Time.seconds(2)) // 允许事件迟到 2 秒.sideOutputLateData(outputTag) // 保存迟到太多的数据.process(new SumProcessWindowFunction());//打印正常的数据result.print();//获取迟到太多的数据DataStream<String> lateDataStream= result.getSideOutput(outputTag).map(new MapFunction<Tuple2<String, Long>, String>() {@Overridepublic String map(Tuple2<String, Long> stringLongTuple2) throws Exception {return "迟到的数据：" + stringLongTuple2.toString();}});lateDataStream.print();env.execute("TimeWindowWordCount");}/*** IN, OUT, KEY, W* IN：输入的数据类型* OUT：输出的数据类型* Key：key的数据类型（在Flink里面，String用Tuple表示）* W：Window的数据类型*/public static class SumProcessWindowFunction extendsProcessWindowFunction<Tuple2<String,Long>,String,Tuple,TimeWindow> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");/*** 当一个window触发计算的时候会调用这个方法* @param tuple key* @param context operator的上下文* @param elements 指定window的所有元素* @param out 用户输出*/@Overridepublic void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,Collector<String> out) {System.out.println("处理时间：" + dateFormat.format(context.currentProcessingTime()));System.out.println("window start time : " + dateFormat.format(context.window().getStart()));List<String> list = new ArrayList<>();for (Tuple2<String, Long> ele : elements) {list.add(ele.toString() + "|" + dateFormat.format(ele.f1));}out.collect(list.toString());System.out.println("window end time  : " + dateFormat.format(context.window().getEnd()));}}private static class EventTimeExtractorimplements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");private long currentMaxEventTime = 0L;private long maxOutOfOrderness = 10000; // 最大允许的乱序时间 10 秒// 拿到每一个事件的 Event Time@Overridepublic long extractTimestamp(Tuple2<String, Long> element,long previousElementTimestamp) {long currentElementEventTime = element.f1;currentMaxEventTime = Math.max(currentMaxEventTime, currentElementEventTime);//打印线程long id = Thread.currentThread().getId();System.out.println("当前线程ID："+id+"event = " + element+ "|" + dateFormat.format(element.f1) // Event Time+ "|" + dateFormat.format(currentMaxEventTime)  // Max Event Time+ "|" + dateFormat.format(getCurrentWatermark().getTimestamp())); // Current Watermarkreturn currentElementEventTime;}@Nullable@Overridepublic Watermark getCurrentWatermark() {/*** WasterMark会周期性的产生，默认就是每隔200毫秒产生一个**         设置 watermark 产生的周期为 1000ms*         env.getConfig().setAutoWatermarkInterval(1000);*/System.out.println("water mark...");return new Watermark(currentMaxEventTime - maxOutOfOrderness);}}
}

WaterMark生成机制

/*** 得到并打印每隔 3 秒钟统计前 3 秒内的相同的 key 的所有的事件* 有条件的产生watermark*/
public class WaterMarkWindowWordCount {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//把并行度设置为2env.setParallelism(2);//步骤一：设置时间类型env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//设置waterMark产生的周期为1senv.getConfig().setAutoWatermarkInterval(1000);// 保存迟到的，会被丢弃的数据OutputTag<Tuple2<String, Long>> outputTag =new OutputTag<Tuple2<String, Long>>("late-date"){};DataStreamSource<String> dataStream = env.socketTextStream("10.148.15.10", 8888);SingleOutputStreamOperator<String> result = dataStream.map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> map(String line) throws Exception {String[] fields = line.split(",");return new Tuple2<>(fields[0], Long.valueOf(fields[1]));}//步骤二：获取数据里面的event Time}).assignTimestampsAndWatermarks(new EventTimeExtractor()).keyBy(0).timeWindow(Time.seconds(3))// .allowedLateness(Time.seconds(2)) // 允许事件迟到 2 秒.sideOutputLateData(outputTag) // 保存迟到太多的数据.process(new SumProcessWindowFunction());//打印正常的数据result.print();//获取迟到太多的数据DataStream<String> lateDataStream= result.getSideOutput(outputTag).map(new MapFunction<Tuple2<String, Long>, String>() {@Overridepublic String map(Tuple2<String, Long> stringLongTuple2) throws Exception {return "迟到的数据：" + stringLongTuple2.toString();}});lateDataStream.print();env.execute("TimeWindowWordCount");}/*** IN, OUT, KEY, W* IN：输入的数据类型* OUT：输出的数据类型* Key：key的数据类型（在Flink里面，String用Tuple表示）* W：Window的数据类型*/public static class SumProcessWindowFunction extendsProcessWindowFunction<Tuple2<String,Long>,String,Tuple,TimeWindow> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");/*** 当一个window触发计算的时候会调用这个方法* @param tuple key* @param context operator的上下文* @param elements 指定window的所有元素* @param out 用户输出*/@Overridepublic void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,Collector<String> out) {System.out.println("处理时间：" + dateFormat.format(context.currentProcessingTime()));System.out.println("window start time : " + dateFormat.format(context.window().getStart()));List<String> list = new ArrayList<>();for (Tuple2<String, Long> ele : elements) {list.add(ele.toString() + "|" + dateFormat.format(ele.f1));}out.collect(list.toString());System.out.println("window end time  : " + dateFormat.format(context.window().getEnd()));}}/*** 按条件产生waterMark*/private static class EventTimeExtractor2implements AssignerWithPunctuatedWatermarks<Tuple2<String, Long>> {@Nullable@Overridepublic Watermark checkAndGetNextWatermark(Tuple2<String, Long> lastElement,long extractedTimestamp) {// 这个方法是每接收到一个事件就会调用// 根据条件产生 watermark ，并不是周期性的产生 watermarkif (lastElement.f0 == "000002") {// 才发送 watermarkreturn new Watermark(lastElement.f1 - 10000);}// 则表示不产生 watermarkreturn null;}@Overridepublic long extractTimestamp(Tuple2<String, Long> element,long previousElementTimestamp) {return element.f1;}}private static class EventTimeExtractorimplements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");private long currentMaxEventTime = 0L;private long maxOutOfOrderness = 10000; // 最大允许的乱序时间 10 秒// 拿到每一个事件的 Event Time@Overridepublic long extractTimestamp(Tuple2<String, Long> element,long previousElementTimestamp) {long currentElementEventTime = element.f1;currentMaxEventTime = Math.max(currentMaxEventTime, currentElementEventTime);long id = Thread.currentThread().getId();System.out.println("当前线程ID："+id+"event = " + element+ "|" + dateFormat.format(element.f1) // Event Time+ "|" + dateFormat.format(currentMaxEventTime)  // Max Event Time+ "|" + dateFormat.format(getCurrentWatermark().getTimestamp())); // Current Watermarkreturn currentElementEventTime;}@Nullable@Overridepublic Watermark getCurrentWatermark() {/*** WasterMark会周期性的产生，默认就是每隔200毫秒产生一个**         设置 watermark 产生的周期为 1000ms*         env.getConfig().setAutoWatermarkInterval(1000);*** 和事件关系不大*    1. watermark 值依赖处理时间的场景*    2. 当有一段时间没有接收到事件，但是仍然需要产生 watermark 的场景*/System.out.println("water mark...");return new Watermark(currentMaxEventTime - maxOutOfOrderness);}}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/508771.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

hbase常见处理方式

hbase常见处理方式

相关依赖 <dependencies><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>1.2.1</version></dependency><dependency><groupId>org.apache.hbase</gro…
阅读更多...
flink连接kafka整合hbase，scala

flink连接kafka整合hbase，scala

解析kafka当中的json格式的数据&#xff0c;入hbase import java.util.Propertiesimport com.alibaba.fastjson.{JSON, JSONObject} import org.apache.flink.api.common.serialization.SimpleStringSchema import org.apache.flink.contrib.streaming.state.RocksDBStateBack…
阅读更多...
sparkStreaming连接kafka整合hbase和redis

sparkStreaming连接kafka整合hbase和redis

sparkStreaming消费kafka数据&#xff0c;并将数据保存到redis和hbase当中去&#xff0c;实现实时 import org.apache.hadoop.hbase.client.{Admin, Connection} import org.apache.hadoop.hbase.{HColumnDescriptor, HTableDescriptor, TableName} import org.apache.kafka.c…
阅读更多...
sparksql一些指标

sparksql一些指标

统计指标 select substr(tb.begin_address_code , 1 ,4) as begin_address_code , count(distinct vehicle_license) as dayVehicleCount from (select begin_address_code , vehicle_license from order where date_format(create_time , yyyy-MM-dd) 2020-02-15 ) tb grou…
阅读更多...
sparkConf常见参数设置

sparkConf常见参数设置

def getSparkConf():SparkConf {val sparkConf: SparkConf new SparkConf().set("spark.driver.cores","4") //设置driver的CPU核数.set("spark.driver.maxResultSize","2g") //设置driver端结果存放的最大容量&#xff0c;这里设置…
阅读更多...
sparkSession常见参数设置

sparkSession常见参数设置

def getSparkSession(sparkConf:SparkConf):SparkSession {val sparkSession: SparkSession SparkSession.builder().config(sparkConf)//调度模式.config("spark.scheduler.mode", "FAIR").config("spark.executor.memoryOverhead", "51…
阅读更多...
关于kafka中acks是否可以为all

关于kafka中acks是否可以为all

kafka源码中有这样一段代码&#xff1a; org.apache.kafka.clients.producer.KafkaProducer private static int parseAcks(String acksString) {try {return acksString.trim().equalsIgnoreCase("all") ? -1 : Integer.parseInt(acksString.trim());} catch (Numb…
阅读更多...
关于统计时间切片标签的一些sql

关于统计时间切片标签的一些sql

------当天付费明细表 DROP TABLE IF EXISTS rpt.tmp_mm_rb_daily_ffmx; create table rpt.tmp_mm_rb_daily_ffmx as select a.* FROM (select c.feemsisdn, c.destmsisdn, c.day, c.price/1000 fee, c.contentid, dc.content_name, c.ordernumber, c.cdrtime, c.createtime, c…
阅读更多...
hadoop 二次开发DatanodeWriteTimeout设置

hadoop 二次开发DatanodeWriteTimeout设置

int getDatanodeWriteTimeout(int numNodes) {return this.dfsClientConf.confTime > 0 ? this.dfsClientConf.confTime 5000 * numNodes : 0;}int getDatanodeReadTimeout(int numNodes) {return this.dfsClientConf.socketTimeout > 0 ? 5000 * numNodes this.dfsC…
阅读更多...
聚类算法

聚类算法

假定样本集 D {X1&#xff0c; 的&#xff0c;…&#xff0c; Xm} 包含 m 个无标记样本&#xff0c; 每个样本 X (X1; X2;… ; Xn) 是一个 n 维特征向量&#xff0c;则聚类算法将样本 集 D 划分为 k 个不相交的簇 {Gl I l 1&#xff0c; 2;… &#xff0c;时&#xff0c;其中…
阅读更多...
k-means均值向量

k-means均值向量

给定样本集 D {Xl) 的&#xff0c;… ,xm}, “k 均值” (k-means )算法针对聚类所 得簇划分 C {C1, C2,…, Ck} 最小化平方误差 ι ELL Ilx 一队IIL il EGi 其中队甘il LEGi X 是簇 q 的均值向量.在一定程度上 刻画了簇内样本围绕簇均值向量的紧密程度&#xff0c; E 值越小则…
阅读更多...
学习向量量化

学习向量量化

与 k 均值算法类似&#xff0c;“学习向量量化” (Learning Vector Quantization&#xff0c;简 称 LVQ)也是试图找到一组原型向量来刻画聚类结构&#xff0c; 但与一般聚类算法不同 的是&#xff0c; LVQ 假设数据样本带有类别标记&#xff0c;学习过程利用样本的这些监督信息…
阅读更多...
k 近邻加权平均

k 近邻加权平均

k 近邻(k-Nearest Neighbor&#xff0c;简称 kNN)学习是一种常用的监督学习方法&#xff0c; 其工作机制非常简单: 给定测试样本?基于某种距离度量找出训练集中与其最 靠近的 k 个训练样本&#xff0c;然后基于这 k 个"邻居"的信息来进行预测. 通常&#xff0c; 在分…
阅读更多...
k 近邻降维

k 近邻降维

k 近邻(k-Nearest Neighbor&#xff0c;简称 kNN)学习是一种常用的监督学习方法&#xff0c; 其工作机制非常简单: 给定测试样本?基于某种距离度量找出训练集中与其最 靠近的 k 个训练样本&#xff0c;然后基于这 k 个"邻居"的信息来进行预测. 通常&#xff0c; 在分…
阅读更多...
维度建模工具

维度建模工具

幵始维度建模工作前&#xff0c;项目组需要理解业务需求&#xff0c;以及作为基础的源数据的实际情况。 通过与、 Ik务代表交流来发现需求&#xff0c;用于理解他们的基于关键性能指标、竞争性商业问题、 决策制定过程、支持分析需求的目标。同时&#xff0c;数据实际情况可以通…
阅读更多...
Cube和Grouping 和Rollup

Cube和Grouping 和Rollup

增强的聚合 Cube和Grouping 和Rollup 这几个分析函数通常用于OLAP中&#xff0c;不能累加&#xff0c;而且需要根据不同维度上钻和下钻的指标统计&#xff0c;比如&#xff0c;分小时、天、月的UV数。 GROUPING SETS 在一个GROUP BY查询中&#xff0c;根据不同的维度组合进行聚…
阅读更多...
常见维度建模错误

常见维度建模错误

需要避免的常见维度建模错误 错误 10: 在事实表中放入文本属性 要从数据仓库事实表中 挑出这些文本属性&#xff0c;并将它们放入维度表中。 错误 9: 限制使用冗长的描述符以节省空间 维度表从几何上看总是比事实表小很多。 错误 8: 将层次划分为多个维度 以用户看来最自然最 有…
阅读更多...
2020-09-21

2020-09-21

columns has 234 elements while hbase.columns.mapping has 92 elements (counting the key if implicit)) 根本原因&#xff1a; 对于4000个字符&#xff0c;hive Metastore中SERDE_PARAMS表中PARAM_VALUE字段的字符限制是此问题的根本原因。 此限制可防止Hive创建具有高列数…
阅读更多...
2020-09-23

2020-09-23

insert into table ads_user_action_convert_day select ‘2019-02-10’, uv.day_count, ua.order_count, cast(ua.order_count/uv.day_count as decimal(10,2)) visitor2order_convert_ratio, ua.payment_count, cast(ua.payment_count/ua.order_count as decimal(10,2)) orde…
阅读更多...
2020-09-28

2020-09-28

Mybatis中#{}和${} 1、#{}将传入的数据都当成一个字符串&#xff0c;会对自动传入的数据加一个引号&#xff08;单引号&#xff1f;双引号&#xff1f;加了引号就对了&#xff09;如&#xff1a; //传入的值是sex order by #{column} 解析后为 order by “sex” //将会出错 2…
阅读更多...
最新文章

Copyright @ 2022~2023