今日目标是深入了解flink的高级api，包括flink的四大基石、窗口操作、时间和水印机制以及状态管理。本文将详细介绍这些关键概念和相关操作，帮助你掌握flink的核心功能。
Flink的四大基石包括：
Checkpoint - 检查点：用于实现分布式一致性，解决数据丢失问题，支持故障恢复。检查点存储的是全局状态，并持久化在HDFS分布式文件系统中。
State - 状态：分为托管状态（Managed State）和原始状态（Raw State）。从数据结构的角度来看，托管状态包括ValueState、ListState、MapState和BroadcastState。
Time - 时间：分为事件时间（EventTime）、摄取时间（IngestionTime）和处理时间（ProcessTime）。
Window - 窗口：用于将无界数据转换为有界数据，支持时间窗口和计数窗口，包括滚动窗口（Tumbling Window）、滑动窗口（Sliding Window）和会话窗口（Session Window）。
窗口操作是Flink处理流数据的关键功能之一。窗口的作用是将动态、无界的数据划定范围，转换为有界、静态的数据进行计算。
为什么需要窗口？
窗口分类：
时间窗口（Time Window）：基于时间进行分类，常见的窗口级别包括一天、一小时、一分钟等。
计数窗口（Count Window）：基于计数进行分类。
如何使用窗口？
时间窗口案例：
需求1：每5秒钟统计一次，最近5秒钟内，各个路口通过红绿灯汽车的数量（基于时间的滚动窗口）。
需求2：每5秒钟统计一次，最近10秒钟内，各个路口通过红绿灯汽车的数量（基于时间的滑动窗口）。
package cn.itcast.flink.basestone;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
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.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
<p>public class WindowDemo01 {
public static void main(String[] args) throws Exception {
//1.env 创建流执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);</p><pre class="brush:php;toolbar:false;"><pre class="brush:php;toolbar:false;"> //2.读取 socket 数据源
DataStreamSource<String> source = env.socketTextStream("192.168.88.161", 9999);
//3.将9,3转为CartInfo(9,3)
DataStream<CartInfo> mapDS = source.map(new MapFunction<String, CartInfo>() {
@Override
public CartInfo map(String value) throws Exception {
String[] kv = value.split(",");
return new CartInfo(kv[0], Integer.parseInt(kv[1]));
}
});
//4.按照 sensorId 分组并划分滚动窗口为5秒，在窗口上求和
// Tumbling(滚动)Processing(处理)TimeWindows(时间窗口)
//需求1:每5秒钟统计一次，最近5秒钟内，各个路口/信号灯通过红绿灯汽车的数量
SingleOutputStreamOperator<CartInfo> result1 = mapDS.keyBy(t -> t.sensorId)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.sum("count");
//需求2:每5秒钟统计一次，最近10秒钟内，各个路口/信号灯通过红绿灯汽车的数量
SingleOutputStreamOperator<CartInfo> result2 = mapDS.keyBy(t -> t.sensorId)
.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))
.sum("count");
//5.打印输出
//result1.print();
result2.print();
//6.execute
env.execute();
}
@Data
@AllArgsConstructor
@NoArgsConstructor
public static class CartInfo {
private String sensorId;//信号灯id
private Integer count;//通过该信号灯的车的数量
}}
计数窗口案例：
需求1：统计在最近5条消息中，各自路口通过的汽车数量，相同的key每出现5次进行统计（基于数量的滚动窗口）。
需求2：统计在最近5条消息中，各自路口通过的汽车数量，相同的key每出现3次进行统计（基于数量的滑动窗口）。
package cn.itcast.flink.basestone;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
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;</p><p>public class CountWindowDemo01 {
public static void main(String[] args) throws Exception {
//1.env 创建流执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);</p><pre class="brush:php;toolbar:false;"><pre class="brush:php;toolbar:false;"> //2.读取 socket 数据源
DataStreamSource<String> source = env.socketTextStream("192.168.88.161", 9999);
//3.将9,3转为CartInfo(9,3)
DataStream<WindowDemo01.CartInfo> mapDS = source.map(new MapFunction<String, WindowDemo01.CartInfo>() {
@Override
public WindowDemo01.CartInfo map(String value) throws Exception {
String[] kv = value.split(",");
return new WindowDemo01.CartInfo(kv[0], Integer.parseInt(kv[1]));
}
});
// * 需求1:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现5次进行统计--基于数量的滚动窗口
// //countWindow(long size, long slide)
SingleOutputStreamOperator<WindowDemo01.CartInfo> result1 = mapDS.keyBy(t -> t.getSensorId())
.countWindow(5)
.sum("count");
// * 需求2:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现3次进行统计--基于数量的滑动窗口
SingleOutputStreamOperator<WindowDemo01.CartInfo> result2 = mapDS.keyBy(t -> t.getSensorId())
.countWindow(5, 3)
.sum("count");
//打印输出
//result1.print();
result2.print();
//执行环境
env.execute();
}
@Data
@AllArgsConstructor
@NoArgsConstructor
public static class CartInfo {
private String sensorId;//信号灯id
private Integer count;//通过该信号灯的车的数量
}}
时间（Time）：
水印机制（Watermark）：
水印机制主要用于解决数据延迟和数据乱序问题。水印（时间戳）等于事件时间减去允许的最大延迟时间。窗口触发的条件是水印时间大于等于窗口的结束时间。
需求：
有订单数据，格式为：(订单ID，用户ID，时间戳/事件时间，订单金额)。要求每隔5秒计算5秒内每个用户的订单总金额，并添加Watermark来解决一定程度上的数据延迟和数据乱序（最多延时3秒）问题。
package cn.itcast.flink.basestone;
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.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.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import java.time.Duration;
import java.util.Random;
import java.util.UUID;</p><p>public class WatermarkDemo01 {
public static void main(String[] args) throws Exception {
//1.env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();</p><pre class="brush:php;toolbar:false;"><pre class="brush:php;toolbar:false;"> //设置属性 ProcessingTime ， 新版本 默认设置 EventTime
//env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//2.Source 创建 Order 类 orderId:String userId:Integer money:Integer eventTime:Long
DataStreamSource<Order> source = env.addSource(new SourceFunction<Order>() {
boolean flag = true;
Random rm = new Random();
@Override
public void run(SourceContext<Order> ctx) throws Exception {
while (flag) {
ctx.collect(new Order(
UUID.randomUUID().toString(),
rm.nextInt(3),
rm.nextInt(101),
//模拟生成 Order 数据 事件时间=当前时间-5秒钟随机*1000
System.currentTimeMillis() - rm.nextInt(5) * 1000
));
Thread.sleep(1000);
}
}
@Override
public void cancel() {
flag = false;
}
});
//3.Transformation
//-告诉Flink要基于事件时间来计算!
//env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//新版本默认就是EventTime
DataStream<Order> result = source.assignTimestampsAndWatermarks(
WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((element, recordTimestamp) -> element.eventTime)
)
//-分配水印机制，最多延迟3秒，告诉Flink数据中的哪一列是事件时间,因为Watermark = 当前最大的事件时间 - 最大允许的延迟时间或乱序时间
//代码走到这里,就已经被添加上Watermark了!接下来就可以进行窗口计算了
//要求每隔5s,计算5秒内(基于时间的滚动窗口)，每个用户的订单总金额
.keyBy(t -> t.userId)
.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;
}}
自定义重写接口实现水印机制：
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.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.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import java.time.Duration;
import java.util.Random;
import java.util.UUID;</p><p>public class WatermarkDemo01 {
public static void main(String[] args) throws Exception {
//1.env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();</p><pre class="brush:php;toolbar:false;"><pre class="brush:php;toolbar:false;"> //设置属性 ProcessingTime ， 新版本 默认设置 EventTime
//env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//2.Source 创建 Order 类 orderId:String userId:Integer money:Integer eventTime:Long
DataStreamSource<Order> source = env.addSource(new SourceFunction<Order>() {
boolean flag = true;
Random rm = new Random();
@Override
public void run(SourceContext<Order> ctx) throws Exception {
while (flag) {
ctx.collect(new Order(
UUID.randomUUID().toString(),
rm.nextInt(3),
rm.nextInt(101),
//模拟生成 Order 数据 事件时间=当前时间-5秒钟随机*1000
System.currentTimeMillis() - rm.nextInt(5) * 1000
));
Thread.sleep(1000);
}
}
@Override
public void cancel() {
flag = false;
}
});
//3.Transformation
//-告诉Flink要基于事件时间来计算!
//env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//新版本默认就是EventTime
DataStream<Order> result = source.assignTimestampsAndWatermarks(
WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((element, recordTimestamp) -> element.eventTime)
)
//-分配水印机制，最多延迟3秒，告诉Flink数据中的哪一列是事件时间,因为Watermark = 当前最大的事件时间 - 最大允许的延迟时间或乱序时间
//代码走到这里,就已经被添加上Watermark了!接下来就可以进行窗口计算了
//要求每隔5s,计算5秒内(基于时间的滚动窗口)，每个用户的订单总金额
.keyBy(t -> t.userId)
.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;
}}
状态是基于key或操作算子的中间结果。Flink的状态分为两种：托管状态（Managed State）和原始状态（Raw State）。托管状态又分为基于key的状态（Keyed State）和基于操作的状态（Operator State）。
Keyed State支持的数据结构包括：
Operator State支持的数据结构包括：
Flink Keyed State 案例：
// Keyed State 案例代码
Flink Operator State 案例：
// Operator State 案例代码
通过以上详细介绍和代码示例，你应该能够更好地理解和应用Flink的高级API，掌握窗口操作、时间和水印机制以及状态管理等关键概念。
以上就是2021年最新最全Flink系列教程__Flink高级API(四)的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
广告Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
388
收藏
