前往小程序，Get更优阅读体验！
立即前往
文档建议反馈控制台
首页
学习
活动
专区
工具
TVP
最新优惠活动
文章/答案/技术大牛
发布
首页
学习
活动
专区
工具
TVP最新优惠活动
返回腾讯云官网
社区首页 >专栏 >2021年大数据Flink（二十三）：???????Watermaker案例演示

2021年大数据Flink（二十三）：???????Watermaker案例演示

作者头像
Lansonli
发布于 2021-10-09 18:01:28
7210
发布于 2021-10-09 18:01:28
举报
文章被收录于专栏：Lansonli技术博客Lansonli技术博客

Watermaker案例演示

需求

有订单数据,格式为: (订单ID，用户ID，时间戳/事件时间，订单金额)

要求每隔5s,计算5秒内，每个用户的订单总金额

并添加Watermaker来解决一定程度上的数据延迟和数据乱序问题。

API

注意:一般我们都是直接使用Flink提供好的BoundedOutOfOrdernessTimestampExtractor

代码实现-1-开发版-掌握

Apache Flink 1.12 Documentation: Generating Watermarks

代码语言：javascript
复制
package cn.it.watermaker;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
?* Author lanson
?* Desc
?* 模拟实时订单数据,格式为: (订单ID，用户ID，订单金额，时间戳/事件时间)
?* 要求每隔5s,计算5秒内(基于时间的滚动窗口)，每个用户的订单总金额
?* 并添加Watermaker来解决一定程度上的数据延迟和数据乱序问题。
?*/
public class WatermakerDemo01_Develop {
????public static void main(String[] args) throws Exception {
????????//1.env
????????StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
????????//2.Source
????????//模拟实时订单数据(数据有延迟和乱序)
????????DataStream<Order> orderDS = env.addSource(new SourceFunction<Order>() {
????????????private boolean flag = true;

????????????@Override
????????????public void run(SourceContext<Order> ctx) throws Exception {
????????????????Random random = new Random();
????????????????while (flag) {
????????????????????String orderId = UUID.randomUUID().toString();
????????????????????int userId = random.nextInt(3);
????????????????????int money = random.nextInt(100);
????????????????????//模拟数据延迟和乱序!
????????????????????long eventTime = System.currentTimeMillis() - random.nextInt(5) * 1000;
????????????????????ctx.collect(new Order(orderId, userId, money, eventTime));

????????????????????TimeUnit.SECONDS.sleep(1);
????????????????}
????????????}

????????????@Override
????????????public void cancel() {
????????????????flag = false;
????????????}
????????});

????????//3.Transformation
????????//-告诉Flink要基于事件时间来计算!
????????//env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//新版本默认就是EventTime
????????//-告诉Flnk数据中的哪一列是事件时间,因为Watermaker = 当前最大的事件时间?- 最大允许的延迟时间或乱序时间
????????/*DataStream<Order> watermakerDS = orderDS.assignTimestampsAndWatermarks(
????????????????new BoundedOutOfOrdernessTimestampExtractor<Order>(Time.seconds(3)) {//最大允许的延迟时间或乱序时间
????????????????????@Override
????????????????????public long extractTimestamp(Order element) {
????????????????????????return element.eventTime;
????????????????????????//指定事件时间是哪一列,Flink底层会自动计算:
????????????????????????//Watermaker = 当前最大的事件时间?- 最大允许的延迟时间或乱序时间
????????????????????}
????????});*/
????????DataStream<Order> watermakerDS = orderDS
????????????????.assignTimestampsAndWatermarks(
????????????????????????WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(3))
????????????????????????????????.withTimestampAssigner((event, timestamp) -> event.getEventTime())
????????????????);

????????//代码走到这里,就已经被添加上Watermaker了!接下来就可以进行窗口计算了
????????//要求每隔5s,计算5秒内(基于时间的滚动窗口)，每个用户的订单总金额
????????DataStream<Order> result = watermakerDS
????????????????.keyBy(Order::getUserId)
????????????????//.timeWindow(Time.seconds(5), Time.seconds(5))
????????????????.window(TumblingEventTimeWindows.of(Time.seconds(5)))
????????????????.sum("money");


????????//4.Sink
????????result.print();

????????//5.execute
????????env.execute();
????}

????@Data
????@AllArgsConstructor
????@NoArgsConstructor
????public static class Order {
????????private String orderId;
????????private Integer userId;
????????private Integer money;
????????private Long eventTime;
????}
}

???????代码实现-2-验证版-了解

代码语言：javascript
复制
package cn.it.watermaker;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
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 java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
?* Author lanson
?* Desc
?* 模拟实时订单数据,格式为: (订单ID，用户ID，订单金额，时间戳/事件时间)
?* 要求每隔5s,计算5秒内(基于时间的滚动窗口)，每个用户的订单总金额
?* 并添加Watermaker来解决一定程度上的数据延迟和数据乱序问题。
?*/
public class WatermakerDemo02_Check {
????public static void main(String[] args) throws Exception {
????????FastDateFormat df = FastDateFormat.getInstance("HH:mm:ss");

????????//1.env
????????StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
????????//2.Source
????????//模拟实时订单数据(数据有延迟和乱序)
????????DataStreamSource<Order> orderDS = env.addSource(new SourceFunction<Order>() {
????????????private boolean flag = true;

????????????@Override
????????????public void run(SourceContext<Order> ctx) throws Exception {
????????????????Random random = new Random();
????????????????while (flag) {
????????????????????String orderId = UUID.randomUUID().toString();
????????????????????int userId = random.nextInt(3);
????????????????????int money = random.nextInt(100);
????????????????????//模拟数据延迟和乱序!
????????????????????long eventTime = System.currentTimeMillis() - random.nextInt(5) * 1000;
????????????????????System.out.println("发送的数据为: "+userId + " : " + df.format(eventTime));
????????????????????ctx.collect(new Order(orderId, userId, money, eventTime));
????????????????????TimeUnit.SECONDS.sleep(1);
????????????????}
????????????}

????????????@Override
????????????public void cancel() {
????????????????flag = false;
????????????}
????????});

????????//3.Transformation
????????/*DataStream<Order> watermakerDS = orderDS
????????????????.assignTimestampsAndWatermarks(
????????????????????????WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(3))
????????????????????????????????.withTimestampAssigner((event, timestamp) -> event.getEventTime())
????????????????);*/

????????//开发中直接使用上面的即可
????????//学习测试时可以自己实现
????????DataStream<Order> watermakerDS = orderDS
????????????????.assignTimestampsAndWatermarks(
????????????????????????new WatermarkStrategy<Order>() {
????????????????????????????@Override
????????????????????????????public WatermarkGenerator<Order> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
????????????????????????????????return new WatermarkGenerator<Order>() {
????????????????????????????????????private int userId = 0;
????????????????????????????????????private long eventTime = 0L;
????????????????????????????????????private final long outOfOrdernessMillis = 3000;
????????????????????????????????????private long maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;

????????????????????????????????????@Override
????????????????????????????????????public void onEvent(Order event, long eventTimestamp, WatermarkOutput output) {
????????????????????????????????????????userId = event.userId;
????????????????????????????????????????eventTime = event.eventTime;
????????????????????????????????????????maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
????????????????????????????????????}

????????????????????????????????????@Override
????????????????????????????????????public void onPeriodicEmit(WatermarkOutput output) {
????????????????????????????????????????//Watermaker = 当前最大事件时间?- 最大允许的延迟时间或乱序时间
????????????????????????????????????????Watermark watermark = new Watermark(maxTimestamp - outOfOrdernessMillis - 1);
????????????????????????????????????????System.out.println("key:" + userId + ",系统时间:" + df.format(System.currentTimeMillis()) + ",事件时间:" + df.format(eventTime) + ",水印时间:" + df.format(watermark.getTimestamp()));
????????????????????????????????????????output.emitWatermark(watermark);
????????????????????????????????????}
????????????????????????????????};
????????????????????????????}
????????????????????????}.withTimestampAssigner((event, timestamp) -> event.getEventTime())
????????????????);


????????//代码走到这里,就已经被添加上Watermaker了!接下来就可以进行窗口计算了
????????//要求每隔5s,计算5秒内(基于时间的滚动窗口)，每个用户的订单总金额
???????/* DataStream<Order> result = watermakerDS
?????????????????.keyBy(Order::getUserId)
????????????????//.timeWindow(Time.seconds(5), Time.seconds(5))
????????????????.window(TumblingEventTimeWindows.of(Time.seconds(5)))
????????????????.sum("money");*/

????????//开发中使用上面的代码进行业务计算即可
????????//学习测试时可以使用下面的代码对数据进行更详细的输出,如输出窗口触发时各个窗口中的数据的事件时间,Watermaker时间
????????DataStream<String> result = watermakerDS
????????????????.keyBy(Order::getUserId)
????????????????.window(TumblingEventTimeWindows.of(Time.seconds(5)))
????????????????//把apply中的函数应用在窗口中的数据上
????????????????//WindowFunction<IN, OUT, KEY, W extends Window>
????????????????.apply(new WindowFunction<Order, String, Integer, TimeWindow>() {
????????????????????@Override
????????????????????public void apply(Integer key, TimeWindow window, Iterable<Order> input, Collector<String> out) throws Exception {
????????????????????????//准备一个集合用来存放属于该窗口的数据的事件时间
????????????????????????List<String> eventTimeList = new ArrayList<>();
????????????????????????for (Order order : input) {
????????????????????????????Long eventTime = order.eventTime;
????????????????????????????eventTimeList.add(df.format(eventTime));
????????????????????????}
????????????????????????String outStr = String.format("key:%s,窗口开始结束:[%s~%s),属于该窗口的事件时间:%s",
????????????????????????????????key.toString(), df.format(window.getStart()), df.format(window.getEnd()), eventTimeList);
????????????????????????out.collect(outStr);
????????????????????}
????????????????});
????????//4.Sink
????????result.print();

????????//5.execute
????????env.execute();
????}

????@Data
????@AllArgsConstructor
????@NoArgsConstructor
????public static class Order {
????????private String orderId;
????????private Integer userId;
????????private Integer money;
????????private Long eventTime;
????}
}
本文参与?腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2021-04-30 ，如有侵权请联系 cloudcommunity@tencent.com 删除
flink
大数据

本文分享自 作者个人站点/博客?前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与?腾讯云自媒体分享计划? ，欢迎热爱写作的你一起参与！

flink
大数据
评论
登录后参与评论
0 条评论
热度
最新
登录 后参与评论
推荐阅读
LV.
文章
0
获赞
0
目录
  • Watermaker案例演示
    • 需求
      • API
        • 代码实现-1-开发版-掌握
          • ???????代码实现-2-验证版-了解
          相关产品与服务
          大数据
          全栈大数据产品，面向海量数据场景，帮助您 “智理无数，心中有数”！
          产品介绍
          精选特惠 用云无忧
          领券

          腾讯云开发者

          扫码关注腾讯云开发者

          扫码关注腾讯云开发者

          领取腾讯云代金券

          热门产品

          热门推荐

          更多推荐

          Copyright ? 2013 - 2024 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有?

          深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059?深公网安备号 44030502008569

          腾讯云计算（北京）有限责任公司 京ICP证150476号 | ?京ICP备11018762号 | 京公网安备号11010802020287

          问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档

          Copyright ? 2013 - 2024 Tencent Cloud.

          All Rights Reserved. 腾讯云 版权所有

          登录 后参与评论
          http://www.vxiaotou.com