1 开始
1.1 概述
00.总结
a.说明
Apache Flink 是一个开源的分布式流处理框架，专为处理无边界和有边界数据流而设计。
作为Apache顶级项目，Flink提供了高性能、高可用、低延迟的流处理引擎。
核心特性包括事件时间处理、精确一次语义、状态管理、窗口操作等。
支持Java、Scala、Python等多种编程语言，提供丰富的API和连接器生态。
b.核心特点
流批统一：同一套API同时处理流数据和批数据，真正做到流批一体
低延迟：实现毫秒级延迟处理，满足实时性要求高的应用场景
高吞吐：单节点每秒可处理数百万事件，集群处理能力可达数十亿事件
状态管理：提供精确一次的状态一致性保证，支持有状态计算
容错机制：基于检查点的轻量级容错机制，快速故障恢复
事件时间：支持事件时间语义，准确处理乱序事件和迟到数据
灵活部署：支持Standalone、YARN、Kubernetes等多种部署模式
01.发展历史
a.起源
2010年：起源于德国柏林工业大学"Stratosphere"研究项目
项目目标是构建一个通用的分布式数据分析平台
2014年：项目捐赠给Apache基金会，并更名为Flink
Flink在德语中意为"快速"，体现了项目的性能特点
b.发展里程碑
2014年：成为Apache顶级项目，发布0.6版本
2015年：发布0.9版本，引入Table API和SQL支持
2016年：发布Flink 1.0.0版本，标志着生产就绪
2017年：发布1.3版本，改进状态管理和容错机制
2018年：发布1.5版本，增强批处理能力
2019年：发布1.9版本，集成PyFlink Python API
2020年：发布1.11版本，支持Kubernetes部署模式
2021年：发布1.14版本，大幅增强SQL功能和性能优化
2022年：发布1.16版本，改进状态后端和运维工具
02.核心概念
a.数据流
无界流：没有开始和结束的连续数据流，需要持续处理
有界流：有固定大小和边界的有限数据集，类似传统批处理
事件时间：数据本身携带的时间戳，反映事件实际发生时间
摄入时间：数据进入Flink系统的时间
处理时间：系统执行处理操作时的机器时间
水位线：用于衡量事件时间进度的机制，处理乱序数据
b.状态管理
算子状态：与算子实例关联的状态，每个并行实例独立维护
键控状态：按照Key分区的状态，相同Key的状态在同一个算子实例中
联合状态：可合并的状态，支持状态扩缩容时的重新分布
广播状态：可广播到所有并行实例的只读状态
TTL状态：支持生存时间配置的状态，自动清理过期数据
c.检查点机制
分布式快照：基于Chandy-Lamport算法的分布式一致性快照
Barrier机制：通过特殊数据记录同步所有算子的状态快照
增量检查点：只保存变化的状态数据，减少存储开销
保存点：手动触发的检查点，用于版本升级和应用迁移
03.应用场景
a.实时数据分析
实时监控仪表板：业务指标实时展示，支持多维分析和钻取
异常检测和告警：基于规则和机器学习的实时异常检测
实时推荐系统：根据用户行为实时调整推荐结果
用户行为分析：实时用户画像和行为模式识别
b.事件驱动应用
金融风控系统：实时交易风险识别和反欺诈处理
物联网数据处理：大规模传感器数据的实时采集和分析
在线机器学习：实时模型训练和预测，支持在线学习
实时竞价广告：根据用户上下文实时计算广告价格
c.ETL数据处理
实时数据清洗：格式转换、数据验证、异常数据处理
数据格式转换：不同数据源之间的格式标准化
实时数据聚合：基于时间窗口的实时统计和聚合计算
数据质量监控：实时数据质量检查和报告
d.其他应用场景
实时日志分析：系统日志的实时分析和问题诊断
网络安全监控：实时威胁检测和网络安全事件响应
供应链优化：实时库存管理和物流优化
游戏数据分析：玩家行为实时分析和游戏体验优化
1.2 架构设计
01.整体架构
a.分层架构
API层：提供DataStream API、Table API、SQL等多种编程接口
DataStream API：核心流处理API，支持Java、Scala
Table API：关系型表操作API，类型安全的表操作
SQL：标准SQL查询接口，降低开发门槛
Python API：PyFlink，支持Python开发流处理应用
Runtime层：分布式流处理引擎，负责作业的执行和管理
作业图优化：对用户作业进行逻辑和物理优化
任务调度：将作业分解为任务并分配到TaskManager
状态管理：提供有状态计算的支持和容错
检查点协调：协调分布式快照的创建和恢复
状态后端：提供可插拔的状态存储和管理
内存状态后端：基于堆内存的状态存储
文件系统状态后端：基于文件系统的持久化状态
RocksDB状态后端：基于RocksDB的高性能状态存储
部署层：支持多种集群管理和资源调度框架
Standalone模式：Flink自带的集群管理
YARN模式：集成Hadoop YARN资源管理
Kubernetes模式：基于K8s容器编排
Mesos模式：集成Apache Mesos资源调度
b.组件关系
JobManager：集群主节点，负责任务调度和管理
接收客户端提交的作业
将作业转换为执行图
向ResourceManager申请资源
调度任务到TaskManager执行
协调检查点和故障恢复
TaskManager：集群工作节点，负责任务执行
在TaskSlot中执行具体的任务
管理任务的内存和网络资源
向JobManager报告心跳和任务状态
执行检查点和状态恢复操作
Client：作业提交客户端
开发和测试作业代码
将作业提交给JobManager
监控作业执行状态
触发保存点和作业取消操作
02.JobManager详解
a.核心职责
作业调度和资源分配
将作业逻辑图转换为物理执行图
根据任务资源需求分配TaskSlot
监控任务执行状态和进度
处理任务失败和重新调度
检查点协调
触发和协调分布式快照创建
收集所有算子的状态快照
处理检查点完成和超时
管理检查点存储和清理
故障恢复
检测TaskManager故障和任务失败
从最近的检查点恢复作业状态
重新调度失败的任务到健康的TaskManager
保证作业的精确一次语义
b.主要组件
Dispatcher：作业接收和分发组件
REST API接口：提供HTTP REST服务
Web Dashboard：Flink Web UI后端服务
作业提交：接收和预处理作业提交请求
会话管理：维护客户端连接和会话状态
ResourceManager：资源管理组件
资源发现：发现和注册可用的TaskManager
资源分配：根据作业需求分配TaskSlot
资源回收：回收作业完成或失败释放的资源
资源监控：监控集群资源使用情况
JobMaster：单个作业的管理器
作业图构建：将用户作业转换为执行图
任务调度：将任务分配到具体的TaskSlot
状态管理：管理作业级别的状态信息
检查点协调：协调作业的检查点操作
03.TaskManager详解
a.核心职责
任务执行
在TaskSlot中执行具体的算子任务
管理任务的生命周期和状态转换
处理数据记录的接收和发送
执行用户定义的业务逻辑
内存管理
托管内存：管理用户代码和算子状态内存
网络内存：分配网络数据传输缓冲区
堆外内存：管理直接内存的使用
内存回收：配合垃圾回收器进行内存回收
网络通信
数据交换：与其他TaskManager进行数据传输
反压处理：根据下游处理速度调节发送速率
心跳机制：向JobManager发送心跳报告状态
故障检测：检测网络分区和节点故障
b.主要组件
TaskSlot：任务执行槽位
资源隔离：提供CPU和内存资源隔离
任务调度：支持多个子任务在同一个Slot中运行
共享组：支持Slot共享组和资源优化
动态分配：支持动态的Slot分配和释放
NetworkBufferPool：网络缓冲池
缓冲区管理：管理网络数据传输缓冲区
内存分配：高效的内存分配和回收机制
反压控制：基于信用基础的流控制机制
压缩支持：可选的数据压缩减少网络传输
MemoryManager：内存管理器
内存分配：统一管理堆内存和堆外内存
内存分段：将内存划分为固定大小的段
页面管理：基于页面的内存分配策略
内存监控：监控内存使用情况和GC行为
04.数据传输
a.网络栈
Netty通信框架
异步IO：基于Netty的高性能异步网络通信
多路复用：支持TCP连接的多路复用
SSL/TLS：支持加密的安全网络传输
自定义序列化：高效的二进制数据序列化
信用基础流控制
信用机制：基于接收端缓冲区大小的流控制
反压传播：从下游向上游传播反压信号
动态调节：根据网络状况动态调整发送速率
公平调度：保证不同任务间的公平网络资源分配
反压机制
自动反压：自动检测和处理下游处理瓶颈
渐进式调节：渐进式调节发送速率避免震荡
跨节点传播：跨TaskManager节点的反压传播
监控指标：提供反压相关的监控指标
b.数据交换策略
点对点交换
直接连接：发送端和接收端直接建立网络连接
负载均衡：在多个接收端间进行负载均衡
故障恢复：支持网络分区和节点故障的恢复
性能优化：针对点对点模式的性能优化
广播交换
一对多：一个发送端向多个接收端广播数据
树形分发：使用树形拓扑减少网络带宽消耗
可靠广播：保证所有接收端都能收到广播数据
动态订阅：支持接收端的动态订阅和取消
重分区交换
数据重分布：根据Key进行数据重新分布
负载均衡：保证数据在分区间的负载均衡
自定义分区：支持自定义的分区策略
倾斜处理：处理数据倾斜问题的特殊策略
1.3 核心特性
01.事件时间处理
a.时间语义
事件时间
定义：数据记录中携带的真实事件发生时间戳
优势：不受处理延迟影响，保证结果的正确性
应用：业务分析、审计、合规性要求高的场景
实现：从数据记录中提取时间戳字段
摄入时间
定义：数据记录进入Flink Source算子时的时间戳
特点：介于事件时间和处理时间之间
适用：无法获取事件时间或事件时间不准确的情况
优势：比处理时间更接近真实事件时间
处理时间
定义：算子执行处理操作时的机器时间
特点：简单易用，延迟最低
适用：对实时性要求极高、允许一定误差的场景
局限：受处理延迟和系统时钟影响
b.水位线机制
水位线概念和原理
定义：衡量事件时间进度的机制，表示"到某个时间为止的所有数据都已经到达"
作用：触发窗口计算、处理乱序数据、控制状态过期
语义：时间戳T的水位线表示后续所有数据的时间戳都小于T
传播：水位线在数据流中随数据一起传播
水位线生成策略
周期性生成：定期根据当前数据的时间戳生成水位线
标记生成：根据特殊的数据记录触发水位线更新
Bounded Out-of-Orderness：允许固定时间范围的乱序
Punctuated Watermarks：基于事件特征的动态水位线
迟到数据处理
侧输出流：将迟到数据输出到侧输出流进行特殊处理
允许延迟：在窗口关闭后保留一段时间等待迟到数据
丢弃策略：直接丢弃迟到数据，保证处理性能
更新策略：使用迟到数据更新之前的计算结果
02.状态管理
a.状态类型
Keyed State（键控状态）
定义：按照Key进行分区的状态，相同Key的数据访问相同状态
类型：ValueState、ListState、MapState、ReducingState等
特点：自动分片、负载均衡、支持扩缩容
应用：基于用户ID的会话状态、基于设备ID的配置状态
Operator State（算子状态）
定义：与算子实例关联的状态，每个并行实例独立维护
类型：ListState、UnionState、BroadcastState
特点：与并行度绑定，需要用户处理分片逻辑
应用：Kafka偏移量管理、Source连接池状态
Broadcast State（广播状态）
定义：广播到所有并行实例的只读状态
特点：低延迟访问、一致性保证、动态更新
应用：规则配置、参数更新、控制信号分发
b.状态后端
MemoryStateBackend（内存状态后端）
特点：状态存储在JVM堆内存中，检查点存储在JobManager内存
优势：访问速度最快、适合开发和调试
局限：受限于JVM堆大小、状态大小受限
适用：状态较小的开发测试环境
FsStateBackend（文件系统状态后端）
特点：状态存储在TaskManager内存，检查点存储在文件系统
优势：支持大状态、持久化存储、适合生产环境
配置：支持本地文件系统、HDFS、S3等分布式存储
适用：中等规模状态的生产环境
RocksDBStateBackend（RocksDB状态后端）
特点：基于RocksDB的本地存储，支持增量检查点
优势：支持超大状态、增量检查点、高性能
局限：访问延迟较高、需要序列化开销
适用：大规模状态的生产环境
c.状态TTL和清理
TTL配置
时间特性：支持访问时间、更新时间、创建时间
状态类型：支持OnRead和OnWrite两种更新策略
可见性：控制TTL过期后是否返回清理状态
清理策略：支持增量和全量清理
状态压缩
增量压缩：定期合并状态变更减少存储开销
前缀压缩：对MapState的前缀键进行压缩
定时清理：定时清理过期和冗余的状态数据
03.检查点机制
a.检查点原理
分布式快照算法
基础算法：基于Chandy-Lamport分布式快照算法
异步执行：检查点操作与数据处理并行执行
增量快照：只保存变化的状态数据减少IO开销
一致性保证：保证所有算子状态的一致性
Barrier机制
Barrier定义：特殊的流控制记录，标识检查点边界
对齐机制：等待所有输入流的Barrier到达
非对齐模式：不等待Barrier，适合低延迟场景
传播过程：Barrier从Source到Sink的完整传播过程
一致性保证
精确一次：保证检查点时机状态的精确一致性
At-least-once：在特定配置下至少一次语义
故障恢复：从检查点恢复保证状态一致性
事务性：结合外部系统实现端到端一致性
b.检查点配置
检查点间隔
配置参数：checkpointing.interval设置检查点频率
调优策略：根据业务需求调整间隔时间
资源消耗：检查点频率与资源消耗的关系
监控指标：检查点延迟、成功率等关键指标
超时配置
超时时间：checkpointing.timeout设置单次检查点超时
超时处理：超时后的处理策略和恢复机制
调优建议：根据数据量和网络状况设置合理的超时
监控告警：设置检查点超时的监控和告警
并发检查点
最大并发数：max-concurrent-checkpoints控制并发检查点数量
资源管理：并发检查点对集群资源的影响
间隔配置：min-pause-between-checkpoints设置最小间隔
性能权衡：并发数量与性能的平衡考虑
保存点机制
手动触发：通过API或CLI手动创建保存点
版本管理：保存点用于应用升级和版本回退
迁移支持：支持跨集群的状态迁移
兼容性：处理Flink版本间的状态兼容性问题
04.窗口操作
a.窗口类型
时间窗口
Tumbling Window（滚动窗口）：固定大小、不重叠的时间窗口
示例：每5秒钟统计一次数据
特点：简单清晰、适合周期性统计
Sliding Window（滑动窗口）：固定大小、可重叠的时间窗口
示例：每1秒钟统计最近5秒钟的数据
特点：更细粒度的统计、资源消耗较大
Session Window（会话窗口）：基于活动间隙的动态窗口
示例：用户会话分析，超时时间30分钟
特点：动态大小、适合用户行为分析
计数窗口
Tumbling Count Window：固定数量的滚动窗口
示例：每1000条记录统计一次
特点：基于数据量而非时间
Sliding Count Window：固定数量的滑动窗口
示例：每100条记录统计最近1000条数据
特点：更频繁的统计、内存开销大
全局窗口
定义：将所有数据分配到一个窗口中
触发条件：必须配合自定义触发器使用
应用场景：全局聚合、条件触发计算
注意事项：内存使用量大、需要谨慎使用
b.窗口触发
时间触发
Event Time Trigger：基于事件时间的水位线触发
Processing Time Trigger：基于处理时间的系统时钟触发
Continuous Event Time Trigger：基于连续事件时间触发
Continuous Processing Time Trigger：基于连续处理时间触发
计数触发
Count Trigger：基于窗口内数据记录数量触发
Delta Trigger：基于数据变化的增量阈值触发
Purging Trigger：触发后清除窗口数据的触发器
自定义触发器
Trigger接口：实现onElement、onProcessingTime、onEventTime方法
触发时机：定义何时触发窗口计算
清除策略：定义何时清理窗口数据
状态管理：触发器内部状态的管理和序列化
c.窗口函数
增量聚合：增量计算，内存占用少
适用场景：sum、min、max、count等简单聚合
限制：只能访问聚合结果，无法访问窗口内所有数据
ProcessWindowFunction
全量聚合：可访问窗口内所有数据和状态
适用场景：复杂计算、需要上下文信息
性能考虑：内存占用大，性能开销高
AggregateFunction
增量聚合的通用接口
支持累加器和结果类型不同
提供更灵活的聚合实现
WindowProcessFunction（Table API/SQL）
表处理窗口函数
支持SQL窗口操作
提供丰富的内置窗口函数
2 核心API
2.1 DataStream API
01.DataStream基础
a.数据源
a.文件数据源
readFile方法：读取文件内容，支持文件监控和连续读取
文件格式：支持文本文件、CSV、JSON等格式
文件监控：监控文件目录变化，实现准实时处理
示例代码：
---
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> textStream = env.readTextFile("file:///path/to/file");
---
b.Socket数据源
socketTextStream方法：从Socket连接读取文本数据
参数配置：主机名、端口号、分隔符、重连策略
适用场景：实时日志传输、测试环境数据模拟
示例代码：
---
DataStream<String> socketStream = env.socketTextStream("localhost", 9999);
---
c.集合数据源
fromCollection方法：从Java集合创建数据流
fromElements方法：从元素列表创建数据流
fromSequence方法：创建数字序列数据流
适用场景：测试、演示、小规模数据处理
示例代码：
---
List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
DataStream<Integer> collectionStream = env.fromCollection(data);
---
d.自定义数据源
SourceFunction接口：实现自定义的数据源逻辑
RichSourceFunction：提供生命周期管理和状态支持
并行源：支持并行度的自定义数据源
示例代码：
---
public class CustomSource extends RichSourceFunction<String> {
private boolean isRunning = true;
@Override
public void run(SourceContext<String> ctx) throws Exception {
while (isRunning) {
String data = generateData();
ctx.collect(data);
Thread.sleep(1000);
}
}
@Override
public void cancel() {
isRunning = false;
}
}
---
b.数据转换
a.Map转换
map算子：一对一的数据转换操作
函数接口：MapFunction接口定义转换逻辑
类型信息：自动推导和手动指定输出类型
示例应用：数据格式转换、字段提取、计算处理
示例代码：
---
DataStream<Integer> numbers = env.fromElements(1, 2, 3, 4, 5);
DataStream<Integer> squared = numbers.map(n -> n * n);
---
b.FlatMap转换
flatMap算子：一对多的数据转换操作
输出机制：通过Collector输出零个或多个结果
典型应用：数据拆分、单词分割、记录展开
示例代码：
---
DataStream<String> sentences = env.fromElements("hello world", "flink streaming");
DataStream<String> words = sentences.flatMap((sentence, out) -> {
for (String word : sentence.split(" ")) {
out.collect(word);
}
});
---
c.Filter过滤
filter算子：根据条件过滤数据记录
布尔判断：返回true保留数据，false丢弃数据
应用场景：数据清洗、条件筛选、异常过滤
示例代码：
---
DataStream<Integer> numbers = env.fromElements(1, 2, 3, 4, 5);
DataStream<Integer> evenNumbers = numbers.filter(n -> n % 2 == 0);
---
d.KeyBy分组
keyBy算子：根据Key对数据流进行分区
分区机制：相同Key的数据被分配到同一个算子实例
状态访问：只有在KeyedStream上才能访问键控状态
示例代码：
---
DataStream<Tuple2<String, Integer>> pairs = env.fromElements(
Tuple2.of("apple", 1),
Tuple2.of("banana", 2),
Tuple2.of("apple", 3)
);
KeyedStream<Tuple2<String, Integer>, String> keyed = pairs.keyBy(t -> t.f0);
---
c.数据输出
a.文件输出
writeAsText方法：将数据流写入文本文件
writeAsCsv方法：以CSV格式写入文件
滚动策略：基于文件大小或时间的文件滚动
示例代码：
---
DataStream<String> result = ...;
result.writeAsText("file:///path/to/output", FileSystem.WriteMode.OVERWRITE);
---
b.Socket输出
writeToSocket方法：向Socket连接发送数据
序列化：支持字符串和自定义序列化格式
实时传输：适合实时数据展示和监控
示例代码：
---
DataStream<String> stream = ...;
stream.writeToSocket("localhost", 9999, new SimpleStringSchema());
---
c.Kafka输出
FlinkKafkaProducer：向Kafka主题写入数据
序列化器：支持多种序列化格式
容错机制：支持精确一次语义保证
示例代码：
---
Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(
"topic-name",
new SimpleStringSchema(),
props
);
stream.addSink(producer);
---
d.自定义输出
SinkFunction接口：实现自定义的数据输出逻辑
RichSinkFunction：提供生命周期管理和状态支持
批量写入：支持批量写入提高性能
示例代码：
---
public class CustomSink extends RichSinkFunction<String> {
@Override
public void invoke(String value, Context context) throws Exception {
// 自定义输出逻辑
writeToExternalSystem(value);
}
}
---
02.高级操作
a.窗口操作
a.时间窗口
timeWindow方法：创建基于时间的滚动窗口
slidingTimeWindow方法：创建基于时间的滑动窗口
事件时间：基于事件时间的水位线触发
处理时间：基于处理时间的系统时钟触发
示例代码：
---
KeyedStream<Tuple2<String, Integer>, String> keyed = ...;
DataStream<Tuple2<String, Integer>> tumbling = keyed
.timeWindow(Time.seconds(5))
.sum(1);
DataStream<Tuple2<String, Integer>> sliding = keyed
.timeWindow(Time.seconds(10), Time.seconds(5))
.sum(1);
---
b.计数窗口
countWindow方法：创建基于计数的滚动窗口
slidingCountWindow方法：创建基于计数的滑动窗口
触发条件：基于窗口内元素数量触发计算
应用场景：固定批量的数据处理
示例代码：
---
KeyedStream<Tuple2<String, Integer>, String> keyed = ...;
DataStream<Tuple2<String, Integer>> countWindow = keyed
.countWindow(100)
.sum(1);
---
c.会话窗口
processingTimeSessionWindows：基于处理时间的会话窗口
eventTimeSessionWindows：基于事件时间的会话窗口
会话间隙：定义会话超时时间
应用场景：用户行为分析、会话统计
示例代码：
---
KeyedStream<Event, String> keyed = ...;
DataStream<String> sessionWindows = keyed
.window(EventTimeSessionWindows.withGap(Time.minutes(10)))
.process(new MyProcessWindowFunction());
---
b.连接操作
a.Join操作
join方法：基于Key连接两个数据流
窗口约束：必须在窗口内进行连接操作
连接类型：内连接、左连接、右连接（通过CoProcessFunction实现）
示例代码：
---
DataStream<Tuple2<String, Integer>> stream1 = ...;
DataStream<Tuple2<String, String>> stream2 = ...;
DataStream<Tuple3<String, Integer, String>> joined = stream1
.join(stream2)
.where(t -> t.f0) // stream1的Key
.equalTo(t -> t.f0) // stream2的Key
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.apply((t1, t2) -> Tuple3.of(t1.f0, t1.f1, t2.f1));
---
b.CoProcessFunction
双流处理：同时处理两个数据流的数据
状态共享：在两个数据流间共享状态信息
定时器：支持基于时间和事件的定时器
应用场景：复杂的事件关联、模式匹配
示例代码：
---
ConnectedStreams<Tuple2<String, Integer>, Tuple2<String, String>> connected =
stream1.connect(stream2);
DataStream<String> result = connected.process(
new CoProcessFunction<
Tuple2<String, Integer>,
Tuple2<String, String>,
String>() {
private ValueState<Integer> state1;
private ValueState<String> state2;
@Override
public void processElement1(Tuple2<String, Integer> value,
Context ctx, Collector<String> out) {
state1.update(value.f1);
// 处理逻辑
}
@Override
public void processElement2(Tuple2<String, String> value,
Context ctx, Collector<String> out) {
state2.update(value.f1);
// 处理逻辑
}
});
---
c.Connect操作
connect方法：连接两个不同类型的数据流
类型兼容：两个流可以是不同的数据类型
共享处理：通过CoProcessFunction或CoMapFunction处理
应用场景：多源数据融合、控制流与数据流结合
示例代码：
---
DataStream<String> controlStream = ...;
DataStream<Tuple2<String, Integer>> dataStream = ...;
ConnectedStreams<Tuple2<String, Integer>, String> connected =
dataStream.connect(controlStream);
DataStream<String> processed = connected.map(
new CoMapFunction<
Tuple2<String, Integer>,
String,
String>() {
@Override
public String map1(Tuple2<String, Integer> value) {
return "Data: " + value.f0 + "," + value.f1;
}
@Override
public String map2(String value) {
return "Control: " + value;
}
});
---
c.状态操作
a.状态访问和更新
ValueState：存储单个值的状态
ListState：存储元素列表的状态
MapState：存储键值对映射的状态
ReducingState：自动进行聚合计算的状态
示例代码：
---
public class MyFunction extends KeyedProcessFunction<String, Event, Result> {
private ValueState<Long> lastTimestamp;
private ListState<Event> eventList;
@Override
public void open(Configuration parameters) {
ValueStateDescriptor<Long> timestampDesc =
new ValueStateDescriptor<>("lastTimestamp", Long.class);
lastTimestamp = getRuntimeContext().getState(timestampDesc);
ListStateDescriptor<Event> listDesc =
new ListStateDescriptor<>("eventList", Event.class);
eventList = getRuntimeContext().getListState(listDesc);
}
@Override
public void processElement(Event value, Context ctx, Collector<Result> out) {
Long lastTs = lastTimestamp.value();
if (lastTs != null) {
// 使用状态进行处理
}
eventList.add(value);
lastTimestamp.update(value.getTimestamp());
}
}
---
b.状态过期策略
StateTtlConfig：配置状态的生存时间
更新策略：OnCreate、OnReadAndWrite、OnReadAndWriteAndGc
状态可见性：NeverReturnExpired、ReturnExpiredIfNotCleanedUp
清理策略：增量清理和全量清理
示例代码：
---
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.hours(1))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.cleanupIncrementally(10, true)
.build();
ValueStateDescriptor<String> descriptor =
new ValueStateDescriptor<>("myState", String.class);
descriptor.enableTimeToLive(ttlConfig);
---
c.状态清理
全量快照：定期的完整状态快照
增量快照：只保存状态变化的部分
本地恢复：从本地磁盘快速恢复状态
RocksDB压缩：定期压缩减少存储空间
监控指标：状态大小、访问频率、垃圾回收等
03.时间操作
a.时间语义设置
a.事件时间配置
setStreamTimeCharacteristic：设置时间语义
事件时间：基于数据记录中的时间戳
水位线：定义水位线生成策略
示例代码：
---
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<Event> events = env.addSource(new EventSource())
.assignTimestampsAndWatermarks(
WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner((event, timestamp) -> event.getTimestamp())
);
---
b.水位线策略
BoundedOutOfOrdernessWatermarks：允许固定时间范围的乱序
TimestampAssigner：自定义时间戳提取逻辑
WatermarkGenerator：实现自定义的水位线生成器
示例代码：
---
// 周期性水位线
WatermarkStrategy<Event> strategy = WatermarkStrategy
.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner((event, timestamp) -> event.getTimestamp());
// 标记水位线
WatermarkStrategy<Event> punctuatedStrategy = WatermarkStrategy
.forGenerator(context -> new PunctuatedWatermarkGenerator())
.withTimestampAssigner((event, timestamp) -> event.getTimestamp());
---
c.时间戳提取器
TimestampAssigner接口：定义时间戳提取逻辑
序列化：时间戳字段的序列化配置
时区处理：处理不同时区的时间戳
示例代码：
---
public class EventTimestampExtractor implements TimestampAssigner<Event> {
@Override
public long extractTimestamp(Event element, long recordTimestamp) {
return element.getEventTime();
}
}
---
b.时间窗口
a.时间窗口定义
EventTimeWindows：基于事件时间的窗口
ProcessingTimeWindows：基于处理时间的窗口
窗口大小：定义窗口的时间长度
滑动间隔：定义滑动窗口的滑动周期
示例代码：
---
// 滚动窗口
WindowedStream<Event, String, TimeWindow> tumbling = keyed
.window(TumblingEventTimeWindows.of(Time.minutes(5)));
// 滑动窗口
WindowedStream<Event, String, TimeWindow> sliding = keyed
.window(SlidingEventTimeWindows.of(
Time.minutes(10), Time.minutes(2)));
// 会话窗口
WindowedStream<Event, String, TimeWindow> session = keyed
.window(EventTimeSessionWindows.withGap(Time.minutes(15)));
---
b.窗口函数
增量聚合函数：ReduceFunction、AggregateFunction
全量窗口函数：ProcessWindowFunction
聚合结果：可访问窗口元数据和上下文信息
示例代码：
---
// ReduceFunction
DataStream<Tuple2<String, Integer>> reduced = windowed
.reduce((t1, t2) -> Tuple2.of(t1.f0, t1.f1 + t2.f1));
// AggregateFunction
DataStream<Double> averaged = windowed.aggregate(
new AggregateFunction<Tuple2<String, Integer>, Tuple2<Integer, Integer>, Double>() {
@Override
public Tuple2<Integer, Integer> createAccumulator() {
return Tuple2.of(0, 0);
}
@Override
public Tuple2<Integer, Integer> add(Tuple2<String, Integer> value,
Tuple2<Integer, Integer> accumulator) {
return Tuple2.of(accumulator.f0 + value.f1, accumulator.f1 + 1);
}
@Override
public Double getResult(Tuple2<Integer, Integer> accumulator) {
return (double) accumulator.f0 / accumulator.f1;
}
@Override
public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> a,
Tuple2<Integer, Integer> b) {
return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
}
});
// ProcessWindowFunction
DataStream<String> processed = windowed.process(
new ProcessWindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {
@Override
public void process(String key,
Context context,
Iterable<Tuple2<String, Integer>> elements,
Collector<String> out) {
int count = 0;
int sum = 0;
for (Tuple2<String, Integer> element : elements) {
count++;
sum += element.f1;
}
TimeWindow window = context.window();
out.collect(String.format(
"Window [%s, %s] Key: %s Count: %d Sum: %d Avg: %.2f",
window.getStart(), window.getEnd(), key, count, sum,
(double) sum / count
));
}
});
---
c.触发器和清理器
Trigger接口：自定义窗口触发条件
Evictor接口：定义窗口数据清理逻辑
事件时间触发器：基于水位线的触发
处理时间触发器：基于系统时间的触发
示例代码：
---
// 自定义触发器
public class CustomTrigger extends Trigger<Event, TimeWindow> {
@Override
public TriggerResult onElement(Event element,
long timestamp,
TimeWindow window,
TriggerContext ctx) {
// 自定义触发逻辑
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onProcessingTime(long time,
TimeWindow window,
TriggerContext ctx) {
// 处理时间触发逻辑
return TriggerResult.FIRE;
}
@Override
public TriggerResult onEventTime(long time,
TimeWindow window,
TriggerContext ctx) {
// 事件时间触发逻辑
return time == window.getEnd() ? TriggerResult.FIRE : TriggerResult.CONTINUE;
}
@Override
public void clear(TimeWindow window, TriggerContext ctx) {
// 清理逻辑
}
}
// 应用自定义触发器
WindowedStream<Event, String, TimeWindow> customWindow = keyed
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.trigger(new CustomTrigger());
---
2.2 Table API & SQL
01.Table API基础
a.表概念
a.动态表和静态表
动态表：表示持续变化的数据流，支持实时查询和更新
静态表：表示有限的数据集，类似传统数据库表
持续查询：动态表上的查询会持续产生结果
表的演变：动态表可以插入、更新、删除数据
b.表模式定义
Schema定义：使用TableDescriptor或DDL语句定义表结构
字段类型：支持基本类型、复合类型、时间类型
约束定义：主键、外键、唯一约束等
水印策略：在表定义中直接设置水位线策略
示例代码：
---
// 通过Schema定义表
Table table = tableEnv.fromDataStream(
stream,
Schema.newBuilder()
.column("userId", DataTypes.STRING())
.column("amount", DataTypes.DECIMAL(10, 2))
.column("eventTime", DataTypes.BIGINT())
.watermark("eventTime", "SOURCE_WATERMARK()")
.build()
);
// 通过DDL定义表
tableEnv.executeSql("""
CREATE TABLE events (
userId STRING,
amount DECIMAL(10, 2),
eventTime TIMESTAMP(3),
WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'events',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'json'
)
""");
---
c.表注册和查询
表注册：通过registerTable或executeSql注册表到表环境
临时视图：创建临时的表视图供查询使用
持久化表：在Catalog中持久化表定义
表查询：使用Table API或SQL查询已注册的表
示例代码：
---
// 注册DataStream为表
tableEnv.createTemporaryView("events", table);
// 查询表
Table result = tableEnv.sqlQuery("""
SELECT userId, SUM(amount) as totalAmount
FROM events
GROUP BY userId
""");
// 创建临时视图
tableEnv.createTemporaryView("filtered_events",
tableEnv.from("events").where($("amount").isGreater(100)));
---
b.表操作
a.选择和过滤
select操作：选择表的特定列或进行计算
filter操作：根据条件过滤表数据
as操作：重命名表的列名
where操作：添加过滤条件
示例代码：
---
// 选择操作
Table selected = table.select($("userId"), $("amount"), $("eventTime"));
// 过滤操作
Table filtered = table.filter($("amount").isGreater(100));
// 条件查询
Table result = table.select($("userId"), $("amount").multiply(0.1).as("tax"))
.where($("eventTime").isGreater(LocalDateTime.now()));
// 复杂条件
Table complexFilter = table.filter(
$("amount").isGreater(100).and($("userId").isEqual("premium_user"))
);
---
b.聚合操作
groupBy分组：按指定字段分组聚合
聚合函数：sum、avg、count、min、max等
分组窗口：时间分组和滑动窗口聚合
分组聚合：group by与聚合函数的组合使用
示例代码：
---
// 简单聚合
Table aggregated = table.groupBy($("userId"))
.select($("userId"), $("amount").sum().as("totalAmount"));
// 多字段聚合
Table multiAgg = table.groupBy($("userId"), $("category"))
.select(
$("userId"),
$("category"),
$("amount").sum().as("total"),
$("amount").avg().as("average"),
$("amount").count().as("count")
);
// 时间窗口聚合
Table windowedAgg = table
.window(Tumble.over(lit(5).minutes()).on($("eventTime")).as("w"))
.groupBy($("w"), $("userId"))
.select($("w").start(), $("userId"), $("amount").sum());
---
c.连接操作
内连接：join操作，返回匹配的数据
左连接：leftJoin操作，保留左表数据
右连接：rightJoin操作，保留右表数据
全连接：fullOuterJoin操作，保留所有数据
示例代码：
---
// 内连接
Table innerJoin = orders.join(customers, $("orders.customerId").isEqual($("customers.id")));
// 左连接
Table leftJoin = orders.leftJoin(customers, $("orders.customerId").isEqual($("customers.id")));
// 带条件的连接
Table conditionalJoin = orders.join(customers,
$("orders.customerId").isEqual($("customers.id"))
.and($("orders.amount").isGreater($("customers.minOrder"))));
// 多表连接
Table multiJoin = orders.join(customers, "customerId")
.join(products, "productId");
---
d.窗口操作
滚动窗口：Tumble窗口，固定大小不重叠
滑动窗口：Slide窗口，固定大小可重叠
会话窗口：Session窗口，基于活动间隙
窗口别名：为窗口定义别名用于后续查询
示例代码：
---
// 滚动窗口聚合
Table tumbleWindow = table
.window(Tumble.over(lit(1).hours()).on($("eventTime")).as("w"))
.groupBy($("w"), $("userId"))
.select($("w").start(), $("userId"), $("amount").sum());
// 滑动窗口聚合
Table slideWindow = table
.window(Slide.over(lit(10).minutes()).every(lit(5).minutes()).on($("eventTime")).as("w"))
.groupBy($("w"), $("category"))
.select($("w").start(), $("category"), $("amount").avg());
// 会话窗口聚合
Table sessionWindow = table
.window(Session.withGap(lit(30).minutes()).on($("eventTime")).as("w"))
.groupBy($("w"), $("userId"))
.select($("w").start(), $w).end(), $("userId"), $("event").count());
---
02.SQL接口
a.SQL语法支持
a.DDL语句
CREATE TABLE：创建表定义，指定连接器和格式
DROP TABLE：删除已存在的表
ALTER TABLE：修改表结构，添加或删除列
CREATE VIEW：创建视图，简化复杂查询
示例代码：
---sql
-- 创建Kafka表
CREATE TABLE kafka_events (
userId STRING,
productId STRING,
amount DECIMAL(10, 2),
eventTime TIMESTAMP(3),
WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'events',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'json',
'json.ignore-parse-errors' = 'true'
);
-- 创建MySQL表
CREATE TABLE mysql_results (
userId STRING,
totalAmount DECIMAL(12, 2),
updateTime TIMESTAMP(3)
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://localhost:3306/flink',
'table-name' = 'user_stats',
'username' = 'flink',
'password' = 'password'
);
-- 创建视图
CREATE VIEW daily_summary AS
SELECT
userId,
DATE(eventTime) as eventDate,
SUM(amount) as dailyAmount,
COUNT(*) as eventCount
FROM kafka_events
GROUP BY userId, DATE(eventTime);
---
b.DML语句
INSERT INTO：向表中插入数据
INSERT OVERWRITE：覆盖写入表数据
UPDATE：更新表中的数据（在可更新表上）
DELETE：删除表中的数据（在可更新表上）
示例代码：
---sql
-- 插入数据
INSERT INTO mysql_results
SELECT userId, SUM(amount) as totalAmount, LOCALTIMESTAMP as updateTime
FROM kafka_events
WHERE eventTime >= CURRENT_DATE
GROUP BY userId;
-- 覆盖写入
INSERT OVERWRITE mysql_results
SELECT userId, SUM(amount) as totalAmount, LOCALTIMESTAMP as updateTime
FROM kafka_events
GROUP BY userId;
-- 批量插入
INSERT INTO kafka_events (userId, productId, amount, eventTime)
VALUES
('user1', 'product1', 100.50, TIMESTAMP '2023-01-01 10:00:00'),
('user2', 'product2', 200.75, TIMESTAMP '2023-01-01 10:01:00');
-- 从查询插入
INSERT INTO kafka_events
SELECT userId, 'default', 50.0, LOCALTIMESTAMP
FROM user_profiles
WHERE isActive = true;
---
c.查询语句
SELECT查询：基础的数据查询语句
WHERE条件：数据过滤条件
GROUP BY分组：数据分组聚合
ORDER BY排序：结果排序
HAVING子句：分组结果过滤
示例代码：
---sql
-- 基础查询
SELECT userId, productId, amount, eventTime
FROM kafka_events
WHERE amount > 100
ORDER BY eventTime DESC
LIMIT 100;
-- 复杂查询
SELECT
e.userId,
c.name as customerName,
p.name as productName,
e.amount,
e.eventTime,
ROW_NUMBER() OVER (PARTITION BY e.userId ORDER BY e.amount DESC) as rank
FROM kafka_events e
JOIN customers c ON e.userId = c.id
JOIN products p ON e.productId = p.id
WHERE e.eventTime >= CURRENT_DATE - INTERVAL '7' DAY
AND c.status = 'ACTIVE'
ORDER BY e.amount DESC;
-- 窗口查询
SELECT
TUMBLE_START(eventTime, INTERVAL '1' HOUR) as windowStart,
TUMBLE_END(eventTime, INTERVAL '1' HOUR) as windowEnd,
userId,
SUM(amount) as hourlyAmount,
COUNT(*) as eventCount
FROM kafka_events
GROUP BY
TUMBLE(eventTime, INTERVAL '1' HOUR),
userId;
-- 递归查询（CTE）
WITH RECURSIVE user_hierarchy AS (
SELECT id, name, managerId, 0 as level
FROM employees
WHERE managerId IS NULL
UNION ALL
SELECT e.id, e.name, e.managerId, h.level + 1
FROM employees e
JOIN user_hierarchy h ON e.managerId = h.id
)
SELECT * FROM user_hierarchy WHERE level <= 3;
---
b.函数支持
a.内置函数
标量函数：字符串、数值、时间、类型转换等函数
聚合函数：sum、avg、count、min、max等
窗口函数：row_number、rank、dense_rank等
表值函数：unnest、explode等
示例代码：
---sql
-- 字符串函数
SELECT
UPPER(name) as upperName,
LOWER(email) as lowerEmail,
CONCAT(firstName, ' ', lastName) as fullName,
SUBSTRING(description, 1, 50) as shortDesc
FROM users;
-- 数值函数
SELECT
ABS(amount) as absoluteAmount,
ROUND(amount, 2) as roundedAmount,
CEIL(amount) as ceilingAmount,
FLOOR(amount) as floorAmount,
MOD(amount, 100) as remainder
FROM transactions;
-- 时间函数
SELECT
eventTime,
DATE_FORMAT(eventTime, 'yyyy-MM-dd HH:mm:ss') as formattedTime,
EXTRACT(HOUR FROM eventTime) as hour,
DATE_DIFF(CURRENT_DATE, DATE(eventTime)) as daysDiff
FROM events;
-- 窗口函数
SELECT
userId,
amount,
ROW_NUMBER() OVER (PARTITION BY userId ORDER BY amount DESC) as rn,
RANK() OVER (PARTITION BY userId ORDER BY amount DESC) as rank,
DENSE_RANK() OVER (PARTITION BY userId ORDER BY amount DESC) as denseRank,
LAG(amount, 1, 0) OVER (PARTITION BY userId ORDER BY eventTime) as prevAmount,
LEAD(amount, 1, 0) OVER (PARTITION BY userId ORDER BY eventTime) as nextAmount
FROM transactions;
---
b.用户自定义函数
标量UDF：一对一转换的函数
表值UDTF：返回多行多列的函数
聚合UDAF：自定义聚合函数
表聚合UDTAF：返回表结果的聚合函数
示例代码：
---
// 标量UDF
public class HashFunction extends ScalarFunction {
public int eval(String str) {
return str.hashCode();
}
public int eval(String str, int seed) {
return str.hashCode() ^ seed;
}
}
// 注册UDF
tableEnv.createTemporaryFunction("hash", HashFunction.class);
// 使用UDF
Table result = tableEnv.sqlQuery(
"SELECT userId, hash(userId) as userHash FROM events"
);
// 表值UDTF
public class SplitFunction extends TableFunction<String> {
public void eval(String str) {
for (String s : str.split(",")) {
collect(s.trim());
}
}
}
// 聚合UDAF
public class WeightedAvg extends AggregateFunction<Double, Tuple2<Double, Integer>> {
@Override
public Tuple2<Double, Integer> createAccumulator() {
return Tuple2.of(0.0, 0);
}
@Override
public Double getValue(Tuple2<Double, Integer> accumulator) {
return accumulator.f0 / accumulator.f1;
}
public void accumulate(Tuple2<Double, Integer> accumulator, Double value, Integer weight) {
accumulator.f0 += value * weight;
accumulator.f1 += weight;
}
}
---
c.聚合函数
内置聚合：sum、avg、count、min、max、first、last等
分组聚合：group by配合聚合函数使用
窗口聚合：窗口内的聚合计算
去重聚合：count distinct、array_agg distinct等
示例代码：
---sql
-- 基础聚合
SELECT
userId,
COUNT(*) as totalEvents,
SUM(amount) as totalAmount,
AVG(amount) as avgAmount,
MIN(amount) as minAmount,
MAX(amount) as maxAmount,
FIRST_VALUE(amount) as firstAmount,
LAST_VALUE(amount) as lastAmount
FROM events
GROUP BY userId;
-- 去重聚合
SELECT
DATE(eventTime) as eventDate,
COUNT(DISTINCT userId) as uniqueUsers,
COUNT(DISTINCT productId) as uniqueProducts,
SUM(DISTINCT amount) as distinctAmountSum
FROM events
GROUP BY DATE(eventTime);
-- 窗口聚合
SELECT
TUMBLE_START(eventTime, INTERVAL '1' HOUR) as hourStart,
COUNT(DISTINCT userId) as hourlyUniqueUsers,
SUM(amount) as hourlyAmount,
AVG(amount) as hourlyAvgAmount
FROM events
GROUP BY TUMBLE(eventTime, INTERVAL '1' HOUR);
-- 多级聚合
SELECT
category,
SUM(totalAmount) as categoryTotal,
AVG(avgAmount) as categoryAvg,
MAX(maxAmount) as categoryMax
FROM (
SELECT
category,
userId,
SUM(amount) as totalAmount,
AVG(amount) as avgAmount,
MAX(amount) as maxAmount
FROM events
GROUP BY category, userId
) subquery
GROUP BY category;
---
d.表值函数
explode：将数组元素展开为多行
unnest：将复合类型展开
lateral view：侧视图连接
自定义表值函数：实现TableFunction接口
示例代码：
---sql
-- explode数组
SELECT
userId,
tag
FROM users
CROSS JOIN UNNEST(tags) AS t(tag);
-- explode map
SELECT
userId,
key,
value
FROM user_properties
CROSS JOIN UNNEST(properties) AS t(key, value);
-- lateral view
SELECT
userId,
order_id,
product_id,
quantity
FROM orders
LATERAL VIEW explode(items) exploded_table AS product_id, quantity;
-- 自定义表值函数使用
SELECT
userId,
word
FROM events
CROSS JOIN TABLE(split_string(description));
---
03.流表转换
a.DataStream转Table
a.流到表的转换
fromDataStream：将DataStream转换为Table
createTemporaryView：创建临时表视图
字段映射：指定字段名和类型映射
类型推导：自动推导或手动指定字段类型
示例代码：
---
// 简单转换
DataStream<Event> eventStream = ...;
Table eventTable = tableEnv.fromDataStream(eventStream);
// 指定字段名
Table eventTable = tableEnv.fromDataStream(
eventStream,
Schema.newBuilder()
.column("userId", DataTypes.STRING())
.column("amount", DataTypes.DECIMAL(10, 2))
.column("eventTime", DataTypes.TIMESTAMP(3))
.build()
);
// 创建临时视图
tableEnv.createTemporaryView("events", eventStream);
// 带时间属性的转换
Table eventTable = tableEnv.fromDataStream(
eventStream,
Schema.newBuilder()
.column("userId", DataTypes.STRING())
.column("amount", DataTypes.DECIMAL(10, 2))
.column("eventTime", DataTypes.BIGINT())
.columnByExpression("procTime", "PROCTIME()")
.watermark("eventTime", "SOURCE_WATERMARK()")
.build()
);
// Tuple类型转换
DataStream<Tuple3<String, Double, Long>> tupleStream = ...;
Table tupleTable = tableEnv.fromDataStream(
tupleStream,
Schema.newBuilder()
.column("f0", "userId") // 重命名字段
.column("f1", "amount")
.column("f2", "eventTime")
.watermark("eventTime", "SOURCE_WATERMARK()")
.build()
);
// POJO类型转换
DataStream<EventPojo> pojoStream = ...;
Table pojoTable = tableEnv.fromDataStream(
pojoStream,
Schema.newBuilder()
.column("userId", DataTypes.STRING())
.column("amount", DataTypes.DECIMAL(10, 2))
.column("eventTime", DataTypes.TIMESTAMP(3))
.column("metadata", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
.build()
);
---
b.表模式定义
基于类型：根据Java类型自动推导字段类型
基于字段名：指定字段名和类型
复杂类型：支持数组、Map、Row等复杂类型
约束定义：定义主键、外键等约束
示例代码：
---
// 基于类型的Schema推导
Table autoSchema = tableEnv.fromDataStream(eventStream);
// 手动Schema定义
Table manualSchema = tableEnv.fromDataStream(
eventStream,
Schema.newBuilder()
.column("userId", DataTypes.STRING().notNull())
.column("amount", DataTypes.DECIMAL(10, 2).notNull())
.column("category", DataTypes.STRING())
.column("eventTime", DataTypes.TIMESTAMP(3))
.column("tags", DataTypes.ARRAY(DataTypes.STRING()))
.column("properties", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
.primaryKey("userId")
.build()
);
// 嵌套类型定义
Table nestedSchema = tableEnv.fromDataStream(
complexStream,
Schema.newBuilder()
.column("user", DataTypes.ROW(
DataTypes.FIELD("id", DataTypes.STRING()),
DataTypes.FIELD("name", DataTypes.STRING()),
DataTypes.FIELD("age", DataTypes.INT())
))
.column("order", DataTypes.ROW(
DataTypes.FIELD("id", DataTypes.STRING()),
DataTypes.FIELD("items", DataTypes.ARRAY(
DataTypes.ROW(
DataTypes.FIELD("productId", DataTypes.STRING()),
DataTypes.FIELD("quantity", DataTypes.INT()),
DataTypes.FIELD("price", DataTypes.DECIMAL(10, 2))
)
))
))
.build()
);
---
c.时间属性定义
处理时间：PROCTIME()定义当前处理时间
事件时间：SOURCE_WATERMARK()定义事件时间和水位线
时间戳列：将BIGINT列转换为TIMESTAMP
时间转换：将字符串时间转换为TIMESTAMP
示例代码：
---
// 处理时间定义
Table withProcTime = tableEnv.fromDataStream(
eventStream,
Schema.newBuilder()
.column("userId", DataTypes.STRING())
.column("amount", DataTypes.DECIMAL(10, 2))
.column("eventTime", DataTypes.BIGINT())
.columnByExpression("procTime", "PROCTIME()")
.build()
);
// 事件时间定义
Table withEventTime = tableEnv.fromDataStream(
eventStream,
Schema.newBuilder()
.column("userId", DataTypes.STRING())
.column("amount", DataTypes.DECIMAL(10, 2))
.column("timestamp", DataTypes.BIGINT())
.columnByExpression("eventTime", "TO_TIMESTAMP(FROM_UNIXTIME(timestamp / 1000))")
.watermark("eventTime", "eventTime - INTERVAL '5' SECOND")
.build()
);
// 混合时间属性
Table withBothTimes = tableEnv.fromDataStream(
eventStream,
Schema.newBuilder()
.column("userId", DataTypes.STRING())
.column("amount", DataTypes.DECIMAL(10, 2))
.column("eventTimestamp", DataTypes.TIMESTAMP(3))
.columnByExpression("procTime", "PROCTIME()")
.watermark("eventTimestamp", "eventTimestamp - INTERVAL '10' SECOND")
.build()
);
// 多级时间转换
Table timeConversion = tableEnv.fromDataStream(
rawStream,
Schema.newBuilder()
.column("userId", DataTypes.STRING())
.column("amount", DataTypes.DECIMAL(10, 2))
.column("eventDate", DataTypes.STRING())
.column("eventTime", DataTypes.STRING())
.columnByExpression("eventDateTime", "TO_TIMESTAMP(CONCAT(eventDate, ' ', eventTime))")
.columnByExpression("procTime", "PROCTIME()")
.watermark("eventDateTime", "eventDateTime - INTERVAL '1' MINUTE")
.build()
);
---
b.Table转DataStream
a.表到流的转换
toDataStream：将Table转换为DataStream
ChangelogMode：定义数据变更模式
toAppendStream：只能插入数据的转换
toRetractStream：支持插入和删除的转换
类型选择：选择合适的转换类型
示例代码：
---
// 转换为DataStream
Table resultTable = tableEnv.sqlQuery("SELECT userId, totalAmount FROM summary");
DataStream<Row> resultStream = tableEnv.toDataStream(resultTable);
// 转换为POJO类型
DataStream<Event> eventStream = tableEnv.toDataStream(
eventTable,
Event.class
);
// 转换为Tuple类型
DataStream<Tuple2<String, Double>> tupleStream = tableEnv.toDataStream(
summaryTable,
DataTypes.ROW(
DataTypes.FIELD("userId", DataTypes.STRING()),
DataTypes.FIELD("totalAmount", DataTypes.DECIMAL(10, 2))
)
);
---
b.更新模式
Append模式：只支持插入操作，数据只能追加
Retract模式：支持插入和删除操作，用boolean标记
Upsert模式：支持插入和更新操作，基于主键更新
示例代码：
---
// Append模式（默认）
Table appendTable = tableEnv.sqlQuery(
"SELECT userId, amount, eventTime FROM events WHERE amount > 100"
);
DataStream<Row> appendStream = tableEnv.toDataStream(appendTable);
// Retract模式
Table retractTable = tableEnv.sqlQuery(
"SELECT userId, SUM(amount) as totalAmount FROM events GROUP BY userId"
);
DataStream<Tuple2<Boolean, Row>> retractStream = tableEnv.toRetractStream(retractTable);
// Upsert模式（需要主键）
Table upsertTable = tableEnv.sqlQuery(
"SELECT userId, FIRST_VALUE(amount) as latestAmount FROM events GROUP BY userId"
);
DataStream<Row> upsertStream = tableEnv.toDataStream(
upsertTable,
Schema.newBuilder()
.column("userId", DataTypes.STRING())
.column("latestAmount", DataTypes.DECIMAL(10, 2))
.primaryKey("userId")
.build()
);
---
c.重分区策略
分区策略：控制数据在算子间的分布
Key分区：基于Key进行数据分区
广播：将数据广播到所有并行实例
重平衡：重新平衡数据分布
示例代码：
---
// Key分区
DataStream<Row> keyPartitioned = resultStream.keyBy(row -> row.getField("userId"));
// 广播
DataStream<Row> broadcasted = configStream.broadcast();
// 重平衡
DataStream<Row> rebalanced = resultStream.rebalance();
// 全局分区
DataStream<Row> global = resultStream.global();
// 自定义分区
DataStream<Row> customPartitioned = resultStream.partitionCustom(
new CustomPartitioner(),
row -> row.getField("userId")
);
---
2.3 连接器
01.Kafka连接器
a.Kafka作为数据源
a.Consumer配置
基本配置：bootstrap.servers、group.id、auto.offset.reset
连接配置：security.protocol、sasl.mechanisms、ssl配置
性能配置：fetch.min.bytes、fetch.max.wait.ms、max.poll.records
可靠性配置：enable.auto.commit、auto.commit.interval.ms
示例代码：
---sql
CREATE TABLE kafka_events (
userId STRING,
productId STRING,
amount DECIMAL(10, 2),
eventTime TIMESTAMP(3),
WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'events',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'flink-consumer-group',
'scan.startup.mode' = 'latest-offset',
'format' = 'json',
'json.fail-on-missing-field' = 'false',
'json.ignore-parse-errors' = 'true'
);
---
b.主题订阅
单主题订阅：指定单一主题名称
正则表达式订阅：使用正则匹配多个主题
主题列表：通过逗号分隔的主题列表
动态发现：自动发现新创建的主题
示例代码：
---
Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "flink-consumer-group");
// 单主题订阅
FlinkKafkaConsumer<Event> consumer = new FlinkKafkaConsumer<>(
"events",
new EventDeserializer(),
props
);
// 正则表达式订阅
FlinkKafkaConsumer<Event> regexConsumer = new FlinkKafkaConsumer<>(
Pattern.compile("events-\\d+"),
new EventDeserializer(),
props
);
// 主题列表订阅
List<String> topics = Arrays.asList("events-1", "events-2", "events-3");
FlinkKafkaConsumer<Event> listConsumer = new FlinkKafkaConsumer<>(
topics,
new EventDeserializer(),
props
);
---
c.偏移量管理
启动模式：earliest-offset、latest-offset、specific-offsets、timestamp
偏移量提交：自动提交和手动提交
检查点集成：与Flink检查点机制集成
偏移量重置：处理偏移量越界问题
示例代码：
---
// 设置启动模式为最早偏移量
consumer.setStartFromEarliest();
// 设置启动模式为最新偏移量
consumer.setStartFromLatest();
// 设置启动时间为特定时间戳
consumer.setStartFromTimestamp(1609459200000L); // 2021-01-01 00:00:00
// 设置特定偏移量
Map<KafkaTopicPartition, Long> specificOffsets = new HashMap<>();
specificOffsets.put(new KafkaTopicPartition("events", 0), 100L);
specificOffsets.put(new KafkaTopicPartition("events", 1), 200L);
consumer.setStartFromSpecificOffsets(specificOffsets);
// 启用检查点集成
env.enableCheckpointing(5000);
consumer.setCommitOffsetsOnCheckpoints(true);
---
d.序列化格式
JSON格式：支持JSON数据的序列化和反序列化
Avro格式：支持Avro Schema演化
CSV格式：支持CSV格式的结构化数据
自定义格式：实现KafkaDeserializationSchema接口
示例代码：
---
// JSON序列化器
public class EventJsonSerializer implements KafkaSerializationSchema<Event> {
@Override
public ProducerRecord<byte[], byte[]> serialize(Event event, Long timestamp) {
try {
ObjectMapper mapper = new ObjectMapper();
byte[] value = mapper.writeValueAsBytes(event);
return new ProducerRecord<>("events", value);
} catch (Exception e) {
throw new RuntimeException("Failed to serialize event", e);
}
}
}
// 自定义反序列化器
public class EventDeserializer implements KafkaDeserializationSchema<Event> {
private ObjectMapper mapper = new ObjectMapper();
@Override
public void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<Event> out) {
try {
Event event = mapper.readValue(record.value(), Event.class);
out.collect(event);
} catch (Exception e) {
// 处理反序列化错误
}
}
@Override
public boolean isEndOfStream(Event nextElement) {
return false;
}
@Override
public TypeInformation<Event> getProducedType() {
return TypeInformation.of(Event.class);
}
}
---
b.Kafka作为数据输出
a.Producer配置
基本配置：bootstrap.servers、acks、retries、batch.size
性能配置：linger.ms、buffer.memory、compression.type
可靠性配置：acks=all、enable.idempotence、transactional.id
安全配置：security.protocol、sasl.mechanisms、ssl配置
示例代码：
---sql
CREATE TABLE kafka_output (
userId STRING,
totalAmount DECIMAL(12, 2),
updateTime TIMESTAMP(3)
) WITH (
'connector' = 'kafka',
'topic' = 'user-aggregates',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.acks' = 'all',
'properties.retries' = '3',
'properties.batch.size' = '16384',
'properties.linger.ms' = '5',
'properties.compression.type' = 'gzip',
'format' = 'json'
);
---
b.分区策略
默认分区：基于消息key的哈希值分区
轮询分区：没有key时使用轮询策略
自定义分区：实现Partitioner接口
分区选择：通过消息头指定特定分区
示例代码：
---
// 自定义分区器
public class CustomPartitioner implements Partitioner<String> {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
// 基于业务逻辑的分区策略
String userId = key.toString();
return Math.abs(userId.hashCode()) % cluster.partitionCountForTopic(topic);
}
@Override
public void close() {}
@Override
public void configure(Map<String, ?> configs) {}
}
// 设置自定义分区器
Properties producerProps = new Properties();
producerProps.setProperty("partitioner.class", CustomPartitioner.class.getName());
FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(
"output-topic",
new SimpleStringSchema(),
producerProps
);
---
c.序列化格式
键值序列化：分别指定key和value的序列化格式
Schema演化：支持数据结构的演化
类型转换：处理类型兼容性问题
性能优化：序列化性能的优化策略
容错机制
精确一次语义：与Flink检查点机制集成
事务性写入：支持Kafka事务
幂等性写入：避免重复数据
错误处理：处理写入失败的策略
示例代码：
---
// 启用精确一次语义
FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(
"output-topic",
new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()),
producerProps,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE
);
// 启用事务性写入
producerProps.setProperty("enable.idempotence", "true");
producerProps.setProperty("transactional.id", "flink-producer-1");
// 自定义错误处理
producer.setFailOnError(false); // 遇到错误时不失败
producer.setLogFailuresOnly(true); // 只记录错误，不重试
---
02.文件系统连接器
a.文件读取
a.批处理模式
一次性读取：读取指定路径下的所有文件
文件过滤：支持文件名和文件大小过滤
并行读取：多个文件并行读取提高性能
监控模式：监控目录变化，实时处理新文件
示例代码：
---sql
-- 批处理模式读取文件
CREATE TABLE batch_input (
userId STRING,
amount DECIMAL(10, 2),
eventTime TIMESTAMP(3)
) WITH (
'connector' = 'filesystem',
'path' = 'file:///path/to/input/data',
'format' = 'csv',
'csv.ignore-parse-errors' = 'true'
);
-- 查询文件数据
SELECT userId, SUM(amount) as totalAmount
FROM batch_input
GROUP BY userId;
---
b.流处理模式
持续监控：持续监控目录下文件变化
文件发现：自动发现新文件并处理
文件完整：确保文件写入完成后再读取
增量处理：只处理新增或修改的文件
示例代码：
---
StreamingFileSink<String> sink = StreamingFileSink
.forRowFormat(new Path("/path/to/output"), new SimpleStringEncoder<String>("UTF-8"))
.withRollingPolicy(
DefaultRollingPolicy.builder()
.withRolloverInterval(TimeUnit.MINUTES.toMinutes(15))
.withInactivityInterval(TimeUnit.MINUTES.toMinutes(5))
.withMaxPartSize(1024 * 1024 * 1024) // 1GB
.build()
)
.build();
stream.addSink(sink);
---
c.文件格式支持
CSV格式：支持不同分隔符和引号配置
JSON格式：支持JSON Lines格式
Parquet格式：列式存储，支持压缩
ORC格式：列式存储，优化分析查询
Avro格式：支持Schema演化
示例代码：
---sql
-- CSV格式
CREATE TABLE csv_table (
id STRING,
name STRING,
age INT
) WITH (
'connector' = 'filesystem',
'path' = 'file:///path/to/csv',
'format' = 'csv',
'csv.field-delimiter' = ',',
'csv.ignore-parse-errors' = 'true'
);
-- JSON格式
CREATE TABLE json_table (
id STRING,
name STRING,
age INT
) WITH (
'connector' = 'filesystem',
'path' = 'file:///path/to/json',
'format' = 'json',
'json.fail-on-missing-field' = 'false'
);
-- Parquet格式
CREATE TABLE parquet_table (
id STRING,
name STRING,
age INT
) WITH (
'connector' = 'filesystem',
'path' = 'file:///path/to/parquet',
'format' = 'parquet'
);
---
d.分区发现
Hive分区：支持Hive风格的分区目录结构
时间分区：基于时间戳的自动分区
自定义分区：支持自定义分区策略
分区裁剪：优化查询性能，只读取相关分区
示例代码：
---sql
-- 分区表定义
CREATE TABLE partitioned_table (
userId STRING,
amount DECIMAL(10, 2),
eventDate STRING,
eventHour STRING
) PARTITIONED BY (eventDate, eventHour) WITH (
'connector' = 'filesystem',
'path' = 'file:///path/to/partitioned',
'format' = 'json'
);
-- 动态分区插入
INSERT INTO partitioned_table
SELECT
userId,
amount,
DATE_FORMAT(eventTime, 'yyyy-MM-dd') as eventDate,
DATE_FORMAT(eventTime, 'HH') as eventHour
FROM events;
-- 分区裁剪查询
SELECT userId, SUM(amount)
FROM partitioned_table
WHERE eventDate = '2023-01-01'
AND eventHour BETWEEN '09' AND '17'
GROUP BY userId;
---
b.文件写入
a.写入格式
文本格式：纯文本或JSON Lines格式
二进制格式：Avro、Parquet、ORC等
压缩格式：支持Gzip、Snappy、LZ4等压缩
分桶格式：支持按字段分桶存储
示例代码：
---
// 文本格式写入
StreamingFileSink<String> textSink = StreamingFileSink
.forRowFormat(new Path("/path/to/output"), new SimpleStringEncoder<>("UTF-8"))
.withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd-HH"))
.build();
// Parquet格式写入
StreamingFileSink<Event> parquetSink = StreamingFileSink
.forBulkFormat(new Path("/path/to/output"), ParquetAvroWriters.forReflectRecord(Event.class))
.withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd"))
.build();
---
b.分桶策略
时间分桶：基于时间戳的目录分桶
哈希分桶：基于字段哈希值的分桶
自定义分桶：实现BucketAssigner接口
分桶大小：控制单个分桶文件的大小
示例代码：
---
// 时间分桶策略
BucketAssigner<Object, String> timeBucketAssigner = new DateTimeBucketAssigner<>(
"yyyy-MM-dd-HH", // 小时级别分桶
ZoneId.systemDefault()
);
// 自定义分桶策略
BucketAssigner<Event, String> customBucketAssigner = new BucketAssigner<Event, String>() {
@Override
public String getBucketId(Event event, Context context) {
// 基于用户ID和日期分桶
return String.format("userId=%s/date=%s",
event.getUserId(),
DateTimeFormatter.ofPattern("yyyy-MM-dd").format(event.getEventTime()));
}
@Override
public Serializers.Serializer<String> getBucketIdSerializer() {
return Serializers.STRING_SERIALIZER;
}
};
StreamingFileSink<Event> bucketedSink = StreamingFileSink
.forRowFormat(new Path("/path/to/output"), new SimpleStringEncoder<>("UTF-8"))
.withBucketAssigner(customBucketAssigner)
.build();
---
c.滚动策略
时间滚动：基于时间间隔的文件滚动
大小滚动：基于文件大小的滚动
不活动滚动：基于写入不活动时间的滚动
自定义滚动：实现RollingPolicy接口
示例代码：
---
// 默认滚动策略
DefaultRollingPolicy<String> rollingPolicy = DefaultRollingPolicy
.builder()
.withRolloverInterval(TimeUnit.MINUTES.toMillis(15)) // 15分钟滚动一次
.withInactivityInterval(TimeUnit.MINUTES.toMillis(5)) // 5分钟无写入则滚动
.withMaxPartSize(1024 * 1024 * 1024) // 1GB大小限制
.build();
// 自定义滚动策略
RollingPolicy<String> customRollingPolicy = new RollingPolicy<String>() {
@Override
public boolean shouldRollOnProcessingTime(PartFileInfo<File, String> partFileState, long currentTime) {
// 基于处理时间的滚动逻辑
return currentTime - partFileState.getCreationTime() > TimeUnit.MINUTES.toMillis(10);
}
@Override
public boolean shouldRollOnEvent(PartFileInfo<File, String> partFileState, String element) {
// 基于事件的滚动逻辑
return false;
}
@Override
public boolean shouldRollOnCheckpoint(PartFileInfo<File, String> partFileState, long checkpointId) {
// 检查点时的滚动逻辑
return checkpointId % 10 == 0; // 每10个检查点滚动一次
}
};
StreamingFileSink<String> rollingSink = StreamingFileSink
.forRowFormat(new Path("/path/to/output"), new SimpleStringEncoder<>("UTF-8"))
.withRollingPolicy(rollingPolicy)
.build();
---
d.压缩选项
Gzip压缩：高压缩比，较慢的压缩速度
Snappy压缩：中等压缩比，快速的压缩速度
LZ4压缩：低压缩比，非常快的压缩速度
自定义压缩：支持自定义压缩算法
示例代码：
---
// 启用压缩的Bulk格式写入
BulkWriter.Factory<Event> bulkWriter = ParquetWriterFactory.forBulkFormat(
new Path("/path/to/output"),
new ParquetWriterFactory<Event>() {
@Override
public BulkWriter<Event> create(FSDataOutputStream out) throws IOException {
return ParquetBulkWriter.forReflectRecord(Event.class, out);
}
}
);
StreamingFileSink<Event> compressedSink = StreamingFileSink
.forBulkFormat(new Path("/path/to/output"), bulkWriter)
.build();
---
03.数据库连接器
a.JDBC连接器
a.数据库连接配置
基本连接：url、username、password、driver
连接池：最大连接数、最小连接数、连接超时
批量操作：批量大小、批量超时
重试机制：连接失败重试次数和间隔
示例代码：
---sql
CREATE TABLE jdbc_sink (
userId STRING,
totalAmount DECIMAL(12, 2),
updateTime TIMESTAMP(3),
PRIMARY KEY (userId) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://localhost:3306/flink_db',
'table-name' = 'user_stats',
'username' = 'flink_user',
'password' = 'flink_password',
'driver' = 'com.mysql.cj.jdbc.Driver',
'connection.pool.size' = '10',
'batch.size' = '1000',
'flush.interval' = '5000'
);
---
b.查询优化
索引利用：确保查询字段有适当索引
分区裁剪：只查询必要的分区
谓词下推：将过滤条件下推到数据库
投影下推：只选择必要的字段
示例代码：
---
// JDBC查询优化
Table jdbcTable = tableEnv.fromValues(
Row.of("SELECT userId, SUM(amount) as totalAmount FROM events WHERE eventTime >= ? GROUP BY userId")
);
// 参数化查询
ParameterizedQuery query = new ParameterizedQuery(
"SELECT * FROM user_stats WHERE update_time > ? AND total_amount > ?",
Arrays.asList(
LocalDateTime.now().minusDays(1),
BigDecimal.valueOf(1000)
)
);
---
c.批量写入
批量插入：使用INSERT语句批量插入数据
批量更新：使用UPDATE语句批量更新数据
批量Upsert：使用MERGE或ON DUPLICATE KEY UPDATE
事务支持：支持事务性批量操作
示例代码：
---
// 批量插入
JdbcSink.sink(
"INSERT INTO user_stats (userId, totalAmount, updateTime) VALUES (?, ?, ?)",
(ps, record) -> {
ps.setString(1, record.getUserId());
ps.setBigDecimal(2, record.getTotalAmount());
ps.setTimestamp(3, Timestamp.valueOf(record.getUpdateTime()));
},
JdbcExecutionOptions.builder()
.withBatchSize(1000)
.withBatchIntervalMs(200)
.withMaxRetries(3)
.build(),
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:mysql://localhost:3306/flink_db")
.withDriverName("com.mysql.cj.jdbc.Driver")
.withUsername("flink_user")
.withPassword("flink_password")
.build()
);
// Upsert写入
JdbcSink.upsertSink(
"INSERT INTO user_stats (userId, totalAmount, updateTime) VALUES (?, ?, ?) " +
"ON DUPLICATE KEY UPDATE totalAmount = VALUES(totalAmount), updateTime = VALUES(updateTime)",
(ps, record) -> {
ps.setString(1, record.getUserId());
ps.setBigDecimal(2, record.getTotalAmount());
ps.setTimestamp(3, Timestamp.valueOf(record.getUpdateTime()));
},
JdbcExecutionOptions.builder().build(),
connectionOptions
);
---
d.事务支持
自动事务：每个批次在一个事务中执行
手动事务：用户控制事务边界
事务隔离：设置事务隔离级别
错误处理：处理事务失败和重试
示例代码：
---
// 事务性写入配置
JdbcExecutionOptions executionOptions = JdbcExecutionOptions.builder()
.withBatchSize(500)
.withBatchIntervalMs(1000)
.withMaxRetries(3)
.build();
// 事务级别配置
Properties connectionProperties = new Properties();
connectionProperties.setProperty("defaultAutoCommit", "false");
connectionProperties.setProperty("transactionIsolation", "TRANSACTION_READ_COMMITTED");
JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:mysql://localhost:3306/flink_db")
.withProperties(connectionProperties)
.build();
---
b.其他数据库
a.Elasticsearch连接器
索引配置：指定索引名称和映射
文档操作：支持索引、更新、删除操作
批量操作：批量写入提高性能
版本控制：支持文档版本控制
示例代码：
---sql
CREATE TABLE es_sink (
userId STRING,
totalAmount DECIMAL(12, 2),
updateTime TIMESTAMP(3)
) WITH (
'connector' = 'elasticsearch-7',
'hosts' = 'http://localhost:9200',
'index' = 'user-stats',
'document-type' = '_doc',
'format' = 'json',
'failure-handler' = 'ignore'
);
---
b.Redis连接器
数据类型：支持String、Hash、List、Set、SortedSet
连接配置：集群模式、哨兵模式配置
命令映射：Flink操作到Redis命令的映射
持久化：支持Redis持久化配置
示例代码：
---
// Redis Sink配置
RedisSink<Tuple2<String, String>> redisSink = new RedisSink<>(
new FlinkJedisPoolConfig.Builder()
.setHost("localhost")
.setPort(6379)
.setDatabase(0)
.build(),
new RedisMapper<Tuple2<String, String>>() {
@Override
public RedisCommandDescription getCommandDescription() {
return new RedisCommandDescription(RedisCommand.HSET, "user_stats");
}
@Override
public String getKeyFromData(Tuple2<String, String> data) {
return data.f0;
}
@Override
public String getValueFromData(Tuple2<String, String> data) {
return data.f1;
}
}
);
stream.addSink(redisSink);
---
c.Cassandra连接器
键空间配置：指定Keyspace和表
查询语句：CQL查询语句配置
批量操作：批量写入优化
一致性级别：设置读写一致性级别
示例代码：
---
// Cassandra Sink
ClusterBuilder clusterBuilder = new ClusterBuilder() {
@Override
protected Cluster buildCluster(Cluster.Builder builder) {
return builder
.addContactPoint("127.0.0.1")
.withPort(9042)
.build();
}
};
CassandraSink<Tuple2<String, Integer>> sink = CassandraSink.addSink(stream)
.setClusterBuilder(clusterBuilder)
.setQuery("INSERT INTO flink.user_stats (user_id, total_amount) VALUES (?, ?)")
.build();
sink.start();
---
d.HBase连接器
表名配置：指定HBase表名和列族
行键设计：RowKey的设计和序列化
列映射：Flink字段到HBase列的映射
版本控制：支持HBase版本控制
示例代码：
---sql
CREATE TABLE hbase_sink (
rowkey STRING,
cf:col1 STRING,
cf:col2 INT
) WITH (
'connector' = 'hbase-2.2',
'table-name' = 'flink_table',
'zookeeper.quorum' = 'localhost:2181'
);
---
3 DataStream API详解
3.1 基础转换操作
01.map映射
a.基本用法
DataStream<Integer> input = ...;
DataStream<Integer> doubled = input.map(x -> x * 2);
一对一转换
应用函数到每个元素
返回新DataStream
b.MapFunction接口
input.map(new MapFunction<Integer, String>() {
@Override
public String map(Integer value) {
return "Number: " + value;
}
});
实现MapFunction接口
类型安全
可以有状态
c.Rich函数
input.map(new RichMapFunction<Integer, String>() {
@Override
public void open(Configuration parameters) {
// 初始化资源
}
@Override
public String map(Integer value) {
return value.toString();
}
@Override
public void close() {
// 清理资源
}
});
访问运行时上下文
生命周期方法
可以访问状态
02.flatMap扁平化映射
a.基本用法
DataStream<String> input = ...;
DataStream<String> words = input.flatMap(
(String line, Collector<String> out) -> {
for (String word : line.split(" ")) {
out.collect(word);
}
}
);
一对多转换
使用Collector收集结果
可以输出0到多个元素
b.FlatMapFunction接口
input.flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String value, Collector<String> out) {
for (String word : value.split(" ")) {
out.collect(word);
}
}
});
实现FlatMapFunction接口
灵活输出
c.应用场景
分词处理
数据解析
过滤+转换
展开嵌套结构
03.filter过滤
a.基本用法
DataStream<Integer> input = ...;
DataStream<Integer> filtered = input.filter(x -> x > 0);
筛选满足条件的元素
返回布尔值
保持元素类型
b.FilterFunction接口
input.filter(new FilterFunction<Integer>() {
@Override
public boolean filter(Integer value) {
return value % 2 == 0;
}
});
实现FilterFunction接口
复杂过滤逻辑
c.组合使用
input.filter(x -> x > 0)
.map(x -> x * 2)
.filter(x -> x < 100);
链式调用
多次过滤
04.keyBy分组
a.基本用法
DataStream<Tuple2<String, Integer>> input = ...;
KeyedStream<Tuple2<String, Integer>, String> keyed =
input.keyBy(value -> value.f0);
按键分组
返回KeyedStream
后续操作按键并行
b.KeySelector
input.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
@Override
public String getKey(Tuple2<String, Integer> value) {
return value.f0;
}
});
自定义键选择
类型安全
c.多字段分组
input.keyBy(value -> Tuple2.of(value.f0, value.f1));
组合键
复合分组
d.注意事项
键必须可序列化
键必须实现hashCode和equals
键不可变
影响分区策略
3.2 聚合操作
01.reduce归约
a.基本用法
KeyedStream<Tuple2<String, Integer>, String> keyed = ...;
DataStream<Tuple2<String, Integer>> reduced =
keyed.reduce((v1, v2) -> Tuple2.of(v1.f0, v1.f1 + v2.f1));
按键归约
滚动聚合
保持状态
b.ReduceFunction接口
keyed.reduce(new ReduceFunction<Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> reduce(
Tuple2<String, Integer> v1,
Tuple2<String, Integer> v2) {
return Tuple2.of(v1.f0, v1.f1 + v2.f1);
}
});
实现ReduceFunction
自定义归约逻辑
c.应用场景
累加求和
最大最小值
计数统计
状态更新
02.sum求和
a.位置索引
KeyedStream<Tuple3<String, Integer, Double>, String> keyed = ...;
keyed.sum(1); // 对第2个字段求和
按位置求和
简化API
b.字段名称
keyed.sum("amount");
按字段名求和
POJO类型
c.滚动求和
每个新元素到达时更新和
保持累计状态
输出当前总和
03.min和max
a.min最小值
keyed.min(1); // 第2个字段的最小值
keyed.min("value");
返回最小值所在的完整元素
只更新指定字段
b.max最大值
keyed.max(1);
keyed.max("value");
返回最大值所在的完整元素
c.minBy和maxBy
keyed.minBy(1); // 返回字段最小的整个元素
keyed.maxBy("value");
返回完整元素
所有字段都是最小/最大时的值
04.aggregate聚合
a.AggregateFunction
keyed.aggregate(new AggregateFunction<
Tuple2<String, Integer>, // 输入类型
Long, // 累加器类型
Double>() { // 输出类型
@Override
public Long createAccumulator() {
return 0L;
}
@Override
public Long add(Tuple2<String, Integer> value, Long acc) {
return acc + value.f1;
}
@Override
public Double getResult(Long acc) {
return acc / 1000.0;
}
@Override
public Long merge(Long a, Long b) {
return a + b;
}
});
自定义聚合逻辑
灵活的累加器
可以改变输出类型
b.应用场景
平均值计算
复杂统计
自定义指标
状态聚合
3.3 窗口操作API
01.滚动窗口
a.时间滚动窗口
keyed.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.sum(1);
固定大小窗口
不重叠
基于事件时间
b.处理时间滚动窗口
keyed.window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
.reduce((v1, v2) -> v1 + v2);
基于处理时间
系统时间
c.计数滚动窗口
keyed.countWindow(100);
固定元素数量
每100个元素一个窗口
02.滑动窗口
a.时间滑动窗口
keyed.window(SlidingEventTimeWindows.of(
Time.minutes(10), // 窗口大小
Time.minutes(5) // 滑动步长
)).sum(1);
窗口可重叠
滑动间隔
b.计数滑动窗口
keyed.countWindow(100, 10);
窗口大小100
每10个元素滑动
元素可能在多个窗口
03.会话窗口
a.事件时间会话窗口
keyed.window(EventTimeSessionWindows.withGap(Time.minutes(10)))
.reduce((v1, v2) -> v1 + v2);
动态窗口大小
间隔超过gap则新窗口
适合用户会话
b.处理时间会话窗口
keyed.window(ProcessingTimeSessionWindows.withGap(Time.seconds(30)))
.sum(1);
基于处理时间
超时检测
c.动态gap
keyed.window(EventTimeSessionWindows.withDynamicGap(
element -> element.getSessionTimeout()
));
每个元素自定义gap
灵活会话定义
04.全局窗口
a.基本用法
keyed.window(GlobalWindows.create())
.trigger(CountTrigger.of(100))
.sum(1);
所有元素在一个窗口
需要自定义触发器
手动控制窗口
b.自定义触发器
需要指定何时触发计算
何时清理窗口
灵活控制
05.窗口函数
a.ReduceFunction
keyed.window(...)
.reduce((v1, v2) -> v1 + v2);
增量聚合
高效
无法访问窗口元数据
b.AggregateFunction
keyed.window(...)
.aggregate(new MyAggregateFunction());
增量聚合
可以改变类型
高效
c.ProcessWindowFunction
keyed.window(...)
.process(new ProcessWindowFunction<
Tuple2<String, Integer>,
String,
String,
TimeWindow>() {
@Override
public void process(
String key,
Context context,
Iterable<Tuple2<String, Integer>> elements,
Collector<String> out) {
int count = 0;
for (Tuple2<String, Integer> e : elements) {
count++;
}
out.collect("Key: " + key + ", Count: " + count);
}
});
访问所有窗口元素
访问窗口元数据
可以输出多个结果
内存占用大
d.增量聚合+ProcessWindowFunction
keyed.window(...)
.aggregate(
new MyAggregateFunction(),
new MyProcessWindowFunction()
);
先增量聚合
再全量处理
兼顾性能和灵活性
06.窗口触发器
a.EventTimeTrigger
默认事件时间触发器
水位线超过窗口结束时间触发
b.ProcessingTimeTrigger
默认处理时间触发器
系统时间超过窗口结束时间触发
c.CountTrigger
CountTrigger.of(100)
元素数量达到阈值触发
d.自定义触发器
class MyTrigger extends Trigger<T, W> {
@Override
public TriggerResult onElement(...) {
// 元素到达时
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onProcessingTime(...) {
// 处理时间定时器触发
return TriggerResult.FIRE;
}
@Override
public TriggerResult onEventTime(...) {
// 事件时间定时器触发
return TriggerResult.FIRE_AND_PURGE;
}
@Override
public void clear(...) {
// 清理状态
}
}
完全自定义触发逻辑
TriggerResult:CONTINUE,FIRE,PURGE,FIRE_AND_PURGE
07.窗口驱逐器
a.CountEvictor
CountEvictor.of(100)
保留最近100个元素
驱逐旧元素
b.TimeEvictor
TimeEvictor.of(Time.seconds(60))
保留最近60秒的元素
c.DeltaEvictor
DeltaEvictor.of(threshold, deltaFunction)
基于差值驱逐
自定义逻辑
3.4 连接操作
01.union合并
a.基本用法
DataStream<Integer> stream1 = ...;
DataStream<Integer> stream2 = ...;
DataStream<Integer> stream3 = ...;
DataStream<Integer> union = stream1.union(stream2, stream3);
合并多个流
相同类型
不去重
b.特点
简单合并
保持顺序
无需分组
可以合并多个流
02.connect连接
a.基本用法
DataStream<Integer> stream1 = ...;
DataStream<String> stream2 = ...;
ConnectedStreams<Integer, String> connected = stream1.connect(stream2);
连接两个不同类型的流
共享状态
协同处理
b.CoMapFunction
connected.map(new CoMapFunction<Integer, String, String>() {
@Override
public String map1(Integer value) {
return "Int: " + value;
}
@Override
public String map2(String value) {
return "Str: " + value;
}
});
分别处理两个流
输出相同类型
c.CoFlatMapFunction
connected.flatMap(new CoFlatMapFunction<Integer, String, String>() {
@Override
public void flatMap1(Integer value, Collector<String> out) {
out.collect("Number: " + value);
}
@Override
public void flatMap2(String value, Collector<String> out) {
for (String word : value.split(" ")) {
out.collect(word);
}
}
});
扁平化处理
灵活输出
03.join连接
a.窗口join
stream1.join(stream2)
.where(value -> value.f0) // stream1的key
.equalTo(value -> value.f0) // stream2的key
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.apply(new JoinFunction<Tuple2<String, Integer>,
Tuple2<String, String>,
String>() {
@Override
public String join(Tuple2<String, Integer> first,
Tuple2<String, String> second) {
return first + " : " + second;
}
});
窗口内join
内连接
相同窗口的元素
b.时间限制
必须在同一窗口
事件时间或处理时间
窗口大小影响结果
c.应用场景
关联两个流
丰富数据
实时join
04.coGroup协同分组
a.基本用法
stream1.coGroup(stream2)
.where(value -> value.f0)
.equalTo(value -> value.f0)
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.apply(new CoGroupFunction<
Tuple2<String, Integer>,
Tuple2<String, String>,
String>() {
@Override
public void coGroup(
Iterable<Tuple2<String, Integer>> first,
Iterable<Tuple2<String, String>> second,
Collector<String> out) {
// 处理两个分组
}
});
窗口内协同分组
访问两个流的所有元素
灵活处理
b.与join区别
coGroup可以处理不匹配的情况
类似外连接
更灵活
c.应用场景
外连接
复杂关联
批量处理
05.intervalJoin间隔join
a.基本用法
keyedStream1.intervalJoin(keyedStream2)
.between(Time.milliseconds(-2), Time.milliseconds(1))
.process(new ProcessJoinFunction<T1, T2, OUT>() {
@Override
public void processElement(T1 left, T2 right,
Context ctx,
Collector<OUT> out) {
out.collect(left + " : " + right);
}
});
时间间隔join
left.timestamp + lowerBound <= right.timestamp <= left.timestamp + upperBound
适合时间相关的join
b.特点
基于事件时间
时间窗口灵活
适合时序数据
内连接语义
3.5 分区和重分布
01.rebalance轮询
a.基本用法
DataStream<Integer> input = ...;
DataStream<Integer> rebalanced = input.rebalance();
轮询分配
负载均衡
打破数据倾斜
b.应用场景
数据倾斜后重分布
均匀分配任务
提高并行度利用率
02.rescale本地轮询
a.基本用法
input.rescale();
本地子集轮询
减少网络传输
上下游并行度倍数关系时高效
b.与rebalance区别
rescale只在本地子集间轮询
rebalance全局轮询
rescale网络开销小
03.shuffle随机
a.基本用法
input.shuffle();
随机分配
完全随机
打破顺序
b.应用场景
需要随机性
打破数据相关性
测试
04.broadcast广播
a.基本用法
input.broadcast();
发送到所有下游任务
每个任务都收到完整数据
适合小数据集
b.应用场景
配置数据广播
规则数据广播
维度表广播
c.Broadcast State
MapStateDescriptor<String, Rule> ruleDescriptor =
new MapStateDescriptor<>("rules", String.class, Rule.class);
BroadcastStream<Rule> ruleBroadcast =
ruleStream.broadcast(ruleDescriptor);
DataStream<Event> result = eventStream
.connect(ruleBroadcast)
.process(new BroadcastProcessFunction<Event, Rule, Result>() {
@Override
public void processElement(Event event, ReadOnlyContext ctx,
Collector<Result> out) {
// 读取广播状态
ReadOnlyBroadcastState<String, Rule> state =
ctx.getBroadcastState(ruleDescriptor);
}
@Override
public void processBroadcastElement(Rule rule, Context ctx,
Collector<Result> out) {
// 更新广播状态
ctx.getBroadcastState(ruleDescriptor).put(rule.getId(), rule);
}
});
广播状态模式
动态规则
配置更新
05.partitionCustom自定义分区
a.基本用法
input.partitionCustom(new Partitioner<String>() {
@Override
public int partition(String key, int numPartitions) {
return Math.abs(key.hashCode()) % numPartitions;
}
}, value -> value.f0);
自定义分区逻辑
完全控制分区
b.应用场景
特殊分区需求
数据倾斜处理
业务相关分区
06.global全局
a.基本用法
input.global();
所有数据发送到第一个分区
单线程处理
保证顺序
b.应用场景
需要全局顺序
单点处理
测试调试
c.注意事项
性能瓶颈
不适合大数据量
谨慎使用
3.6 迭代操作
01.iterate迭代
a.基本用法
IterativeStream<Integer> iteration = input.iterate();
DataStream<Integer> iterationBody = iteration
.map(x -> x + 1)
.filter(x -> x < 100);
DataStream<Integer> feedback = iterationBody.filter(x -> x < 50);
iteration.closeWith(feedback);
DataStream<Integer> output = iterationBody.filter(x -> x >= 50);
迭代流
反馈循环
收敛条件
b.应用场景
机器学习迭代
图算法
收敛计算
c.注意事项
需要收敛条件
避免无限循环
性能考虑
02.iterateStream流迭代
a.基本用法
DataStream<Tuple2<Integer, Integer>> input = ...;
IterativeStream<Tuple2<Integer, Integer>> iteration =
input.iterate(5000); // 超时时间
DataStream<Tuple2<Integer, Integer>> iterationBody = iteration
.map(value -> Tuple2.of(value.f0, value.f1 + 1));
DataStream<Tuple2<Integer, Integer>> feedback =
iterationBody.filter(value -> value.f1 < 10);
iteration.closeWith(feedback);
DataStream<Tuple2<Integer, Integer>> output =
iterationBody.filter(value -> value.f1 >= 10);
设置超时
防止死锁
3.7 异步IO操作
01.AsyncDataStream
a.基本用法
DataStream<String> input = ...;
DataStream<Tuple2<String, String>> result =
AsyncDataStream.unorderedWait(
input,
new AsyncDatabaseRequest(),
1000, // 超时时间
TimeUnit.MILLISECONDS,
100 // 最大并发请求数
);
异步IO
提高吞吐量
非阻塞
b.AsyncFunction实现
class AsyncDatabaseRequest extends RichAsyncFunction<String, Tuple2<String, String>> {
private transient DatabaseClient client;
@Override
public void open(Configuration parameters) {
client = new DatabaseClient();
}
@Override
public void asyncInvoke(String input, ResultFuture<Tuple2<String, String>> resultFuture) {
CompletableFuture<String> future = client.query(input);
future.whenComplete((result, error) -> {
if (error != null) {
resultFuture.completeExceptionally(error);
} else {
resultFuture.complete(Collections.singleton(
Tuple2.of(input, result)));
}
});
}
@Override
public void close() {
client.close();
}
}
实现AsyncFunction
异步查询
CompletableFuture
c.有序和无序
AsyncDataStream.orderedWait() // 保持顺序
AsyncDataStream.unorderedWait() // 不保证顺序,吞吐量更高
根据需求选择
02.应用场景
a.数据库查询
异步查询数据库
丰富流数据
提高吞吐量
b.外部服务调用
HTTP请求
RPC调用
缓存查询
c.性能优化
减少阻塞
提高并发
充分利用资源
3.8 Side Output侧输出流
01.侧输出基本用法
a.定义OutputTag
OutputTag<String> lateDataTag = new OutputTag<String>("late-data"){};
定义侧输出标签
泛型类型
唯一标识
b.ProcessFunction中使用
input.process(new ProcessFunction<Event, Event>() {
@Override
public void processElement(Event value, Context ctx,
Collector<Event> out) {
if (value.isLate()) {
ctx.output(lateDataTag, "Late: " + value);
} else {
out.collect(value);
}
}
});
ctx.output输出到侧输出
主输出用out.collect
c.获取侧输出流
DataStream<Event> mainStream = input.process(...);
DataStream<String> lateStream = mainStream.getSideOutput(lateDataTag);
获取侧输出流
独立处理
02.窗口侧输出
a.迟到数据
OutputTag<Event> lateTag = new OutputTag<Event>("late"){};
DataStream<Result> result = input
.keyBy(...)
.window(...)
.sideOutputLateData(lateTag)
.process(...);
DataStream<Event> lateData = result.getSideOutput(lateTag);
处理迟到数据
不丢弃
单独处理
b.应用场景
迟到数据处理
异常数据分流
多路输出
数据分类
03.侧输出最佳实践
a.数据分流
根据条件分流
主流和侧流
独立处理逻辑
b.异常处理
异常数据输出到侧流
不影响主流程
单独监控和处理
c.多路输出
一个输入多个输出
不同类型数据
灵活路由
4 Table API & SQL详解
4.1 Table API基础
01.创建TableEnvironment
a.流式环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
流式Table环境
基于DataStream
b.批式环境
ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment batchTableEnv = BatchTableEnvironment.create(batchEnv);
批式Table环境
基于DataSet
c.统一环境
EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.inStreamingMode()
.build();
TableEnvironment tableEnv = TableEnvironment.create(settings);
统一API
流批一体
02.创建表
a.从DataStream创建
DataStream<Tuple2<String, Integer>> stream = ...;
Table table = tableEnv.fromDataStream(stream, $("name"), $("age"));
从流创建表
指定字段名
b.从文件创建
tableEnv.executeSql(
"CREATE TABLE MyTable (" +
" name STRING," +
" age INT" +
") WITH (" +
" 'connector' = 'filesystem'," +
" 'path' = 'file:///path/to/file'," +
" 'format' = 'csv'" +
")"
);
DDL创建表
指定connector
c.注册表
tableEnv.createTemporaryView("MyTable", table);
注册临时视图
SQL查询使用
03.Table查询
a.select选择
Table result = table.select($("name"), $("age"));
选择字段
投影操作
b.where过滤
Table result = table.where($("age").isGreater(18));
条件过滤
谓词表达式
c.groupBy分组
Table result = table
.groupBy($("name"))
.select($("name"), $("age").sum().as("total"));
分组聚合
聚合函数
d.orderBy排序
Table result = table.orderBy($("age").desc());
排序
升序降序
04.Table转换
a.转DataStream
DataStream<Row> stream = tableEnv.toDataStream(table);
表转流
Row类型
b.转换为特定类型
DataStream<Tuple2<String, Integer>> stream =
tableEnv.toDataStream(table,
TypeInformation.of(new TypeHint<Tuple2<String, Integer>>(){}));
指定类型
类型安全
c.转Changelog流
DataStream<Row> changelogStream = tableEnv.toChangelogStream(table);
变更日志流
包含操作类型
4.2 SQL查询语法
01.基本查询
a.SELECT语句
Table result = tableEnv.sqlQuery(
"SELECT name, age FROM MyTable WHERE age > 18"
);
标准SQL语法
返回Table对象
b.INSERT语句
tableEnv.executeSql(
"INSERT INTO TargetTable SELECT * FROM SourceTable"
);
插入数据
执行SQL
c.CREATE语句
tableEnv.executeSql(
"CREATE TABLE Orders (" +
" order_id BIGINT," +
" product STRING," +
" amount DECIMAL(10,2)" +
") WITH (...)"
);
创建表
DDL语句
02.WHERE条件
a.比较运算符
SELECT * FROM Orders WHERE amount > 100
SELECT * FROM Orders WHERE product = 'Apple'
等于、大于、小于
b.逻辑运算符
SELECT * FROM Orders WHERE amount > 100 AND product LIKE 'A%'
AND、OR、NOT
组合条件
c.IN和BETWEEN
SELECT * FROM Orders WHERE product IN ('Apple', 'Banana')
SELECT * FROM Orders WHERE amount BETWEEN 100 AND 500
范围查询
03.GROUP BY分组
a.基本分组
SELECT product, COUNT(*) as cnt, SUM(amount) as total
FROM Orders
GROUP BY product
分组聚合
聚合函数
b.HAVING过滤
SELECT product, SUM(amount) as total
FROM Orders
GROUP BY product
HAVING SUM(amount) > 1000
分组后过滤
聚合条件
c.多字段分组
SELECT product, category, COUNT(*) as cnt
FROM Orders
GROUP BY product, category
组合分组
04.JOIN连接
a.INNER JOIN
SELECT o.order_id, o.product, u.name
FROM Orders o
INNER JOIN Users u ON o.user_id = u.user_id
内连接
匹配记录
b.LEFT JOIN
SELECT o.order_id, o.product, u.name
FROM Orders o
LEFT JOIN Users u ON o.user_id = u.user_id
左外连接
保留左表所有记录
c.RIGHT JOIN
SELECT o.order_id, o.product, u.name
FROM Orders o
RIGHT JOIN Users u ON o.user_id = u.user_id
右外连接
d.FULL OUTER JOIN
SELECT o.order_id, u.name
FROM Orders o
FULL OUTER JOIN Users u ON o.user_id = u.user_id
全外连接
保留两表所有记录
05.子查询
a.标量子查询
SELECT * FROM Orders
WHERE amount > (SELECT AVG(amount) FROM Orders)
返回单个值
比较运算
b.IN子查询
SELECT * FROM Orders
WHERE product IN (SELECT product FROM TopProducts)
集合判断
c.EXISTS子查询
SELECT * FROM Users u
WHERE EXISTS (SELECT 1 FROM Orders o WHERE o.user_id = u.user_id)
存在性判断
06.UNION合并
a.UNION
SELECT product FROM Orders
UNION
SELECT product FROM Returns
合并结果
去重
b.UNION ALL
SELECT product FROM Orders
UNION ALL
SELECT product FROM Returns
合并结果
不去重
4.3 时间属性
01.Processing Time处理时间
a.定义处理时间
DataStream<Tuple2<String, Long>> stream = ...;
Table table = tableEnv.fromDataStream(
stream,
$("name"),
$("timestamp"),
$("proc_time").proctime()
);
定义处理时间属性
系统时间
b.SQL中定义
CREATE TABLE Orders (
order_id BIGINT,
product STRING,
proc_time AS PROCTIME()
) WITH (...)
PROCTIME()函数
自动生成
c.应用场景
简单时间窗口
不需要精确时间
低延迟要求
02.Event Time事件时间
a.定义事件时间
DataStream<Event> stream = ...;
Table table = tableEnv.fromDataStream(
stream,
$("name"),
$("timestamp").rowtime()
);
指定事件时间字段
rowtime属性
b.SQL中定义
CREATE TABLE Orders (
order_id BIGINT,
product STRING,
order_time TIMESTAMP(3),
WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (...)
WATERMARK定义
延迟容忍
c.水位线策略
order_time - INTERVAL '5' SECOND
固定延迟
order_time - INTERVAL '0' SECOND
无延迟
d.应用场景
精确时间语义
乱序处理
迟到数据处理
03.时间属性使用
a.窗口聚合
SELECT
TUMBLE_START(order_time, INTERVAL '1' HOUR) as window_start,
product,
SUM(amount) as total
FROM Orders
GROUP BY TUMBLE(order_time, INTERVAL '1' HOUR), product
时间窗口
基于时间属性
b.时间过滤
SELECT * FROM Orders
WHERE order_time > CURRENT_TIMESTAMP - INTERVAL '1' DAY
时间范围过滤
c.时间Join
SELECT *
FROM Orders o
JOIN Shipments s
ON o.order_id = s.order_id
AND s.ship_time BETWEEN o.order_time AND o.order_time + INTERVAL '2' DAY
时间间隔Join
4.4 窗口TVF
01.Tumble滚动窗口
a.基本语法
SELECT
window_start,
window_end,
product,
SUM(amount) as total
FROM TABLE(
TUMBLE(TABLE Orders, DESCRIPTOR(order_time), INTERVAL '1' HOUR)
)
GROUP BY window_start, window_end, product
滚动窗口TVF
固定大小
不重叠
b.窗口属性
window_start:窗口开始时间
window_end:窗口结束时间
window_time:窗口时间属性
c.应用场景
固定时间统计
小时级别聚合
日报表
02.Hop滑动窗口
a.基本语法
SELECT
window_start,
window_end,
product,
SUM(amount) as total
FROM TABLE(
HOP(TABLE Orders,
DESCRIPTOR(order_time),
INTERVAL '5' MINUTE, -- 滑动步长
INTERVAL '10' MINUTE) -- 窗口大小
)
GROUP BY window_start, window_end, product
滑动窗口TVF
可重叠
b.窗口重叠
窗口大小 > 滑动步长
元素可能在多个窗口
适合移动平均
c.应用场景
移动统计
趋势分析
实时监控
03.Cumulate累积窗口
a.基本语法
SELECT
window_start,
window_end,
product,
SUM(amount) as total
FROM TABLE(
CUMULATE(TABLE Orders,
DESCRIPTOR(order_time),
INTERVAL '1' HOUR, -- 累积步长
INTERVAL '1' DAY) -- 最大窗口
)
GROUP BY window_start, window_end, product
累积窗口TVF
逐步扩大
b.窗口特点
窗口从0开始累积
每个步长触发一次
最终达到最大窗口
c.应用场景
日内累积统计
UV/PV统计
实时大屏
04.Session会话窗口
a.基本语法
SELECT
window_start,
window_end,
user_id,
COUNT(*) as event_count
FROM TABLE(
SESSION(TABLE UserEvents,
DESCRIPTOR(event_time),
INTERVAL '30' MINUTE) -- 会话间隔
)
GROUP BY window_start, window_end, user_id
会话窗口TVF
动态窗口
b.窗口特点
间隔超过gap则新窗口
窗口大小不固定
适合用户行为分析
c.应用场景
用户会话分析
行为序列分析
超时检测
4.5 聚合函数
01.内置聚合函数
a.基本聚合
COUNT(*):计数
COUNT(DISTINCT col):去重计数
SUM(col):求和
AVG(col):平均值
MAX(col):最大值
MIN(col):最小值
标准SQL聚合
b.字符串聚合
LISTAGG(col, ','):字符串连接
STRING_AGG(col, ','):字符串聚合
拼接字符串
c.统计函数
STDDEV_POP(col):总体标准差
STDDEV_SAMP(col):样本标准差
VAR_POP(col):总体方差
VAR_SAMP(col):样本方差
统计分析
02.自定义标量函数UDF
a.定义UDF
public class SubstringFunction extends ScalarFunction {
public String eval(String s, Integer begin, Integer end) {
return s.substring(begin, end);
}
}
继承ScalarFunction
实现eval方法
b.注册UDF
tableEnv.createTemporarySystemFunction("SubstringFunc", SubstringFunction.class);
注册函数
SQL中使用
c.使用UDF
SELECT SubstringFunc(name, 0, 3) FROM Users
Table API:table.select(call("SubstringFunc", $("name"), 0, 3))
调用自定义函数
03.自定义聚合函数UDAF
a.定义UDAF
public class WeightedAvg extends AggregateFunction<Long, WeightedAvgAccumulator> {
public static class WeightedAvgAccumulator {
public long sum = 0;
public int count = 0;
}
@Override
public WeightedAvgAccumulator createAccumulator() {
return new WeightedAvgAccumulator();
}
public void accumulate(WeightedAvgAccumulator acc, long value) {
acc.sum += value;
acc.count += 1;
}
@Override
public Long getValue(WeightedAvgAccumulator acc) {
if (acc.count == 0) return null;
return acc.sum / acc.count;
}
}
继承AggregateFunction
定义累加器
实现accumulate和getValue
b.注册UDAF
tableEnv.createTemporarySystemFunction("WeightedAvg", WeightedAvg.class);
注册聚合函数
c.使用UDAF
SELECT product, WeightedAvg(amount) FROM Orders GROUP BY product
分组聚合使用
04.自定义表函数UDTF
a.定义UDTF
public class SplitFunction extends TableFunction<Tuple2<String, Integer>> {
public void eval(String str) {
for (String s : str.split(" ")) {
collect(Tuple2.of(s, s.length()));
}
}
}
继承TableFunction
实现eval方法
使用collect输出
b.注册UDTF
tableEnv.createTemporarySystemFunction("Split", SplitFunction.class);
注册表函数
c.使用UDTF
SELECT name, word, length
FROM Users, LATERAL TABLE(Split(name)) AS T(word, length)
LATERAL TABLE展开
一行变多行
05.窗口聚合函数
a.窗口内聚合
SELECT
window_start,
product,
COUNT(*) as cnt,
SUM(amount) as total,
AVG(amount) as avg_amount
FROM TABLE(TUMBLE(...))
GROUP BY window_start, window_end, product
窗口内标准聚合
b.窗口排名
SELECT *
FROM (
SELECT *,
ROW_NUMBER() OVER (
PARTITION BY window_start, window_end
ORDER BY amount DESC
) as row_num
FROM TABLE(TUMBLE(...))
)
WHERE row_num <= 10
窗口内TopN
排名函数
4.6 Catalog和元数据
01.Catalog概念
a.定义
元数据管理
表、视图、函数的注册中心
支持多种Catalog实现
b.内置Catalog
GenericInMemoryCatalog:内存Catalog
HiveCatalog:Hive元数据
JdbcCatalog:JDBC元数据
c.Catalog层次
Catalog -> Database -> Table/View/Function
三层结构
02.使用Catalog
a.创建Catalog
Catalog catalog = new GenericInMemoryCatalog("my_catalog");
tableEnv.registerCatalog("my_catalog", catalog);
注册Catalog
b.切换Catalog
tableEnv.useCatalog("my_catalog");
切换当前Catalog
c.切换Database
tableEnv.useDatabase("my_database");
切换当前数据库
03.HiveCatalog集成
a.配置HiveCatalog
String name = "myhive";
String defaultDatabase = "default";
String hiveConfDir = "/path/to/hive/conf";
HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
tableEnv.registerCatalog("myhive", hive);
tableEnv.useCatalog("myhive");
集成Hive元数据
读写Hive表
b.读取Hive表
Table hiveTable = tableEnv.from("myhive.default.my_table");
直接读取Hive表
使用Hive SerDe
c.写入Hive表
tableEnv.executeSql(
"INSERT INTO myhive.default.my_table SELECT * FROM source_table"
);
写入Hive表
分区支持
04.元数据操作
a.列出Catalog
String[] catalogs = tableEnv.listCatalogs();
获取所有Catalog
b.列出数据库
String[] databases = tableEnv.listDatabases();
获取所有数据库
c.列出表
String[] tables = tableEnv.listTables();
获取所有表
d.获取表Schema
TableSchema schema = tableEnv.from("my_table").getSchema();
获取表结构
4.7 与DataStream互转
01.DataStream转Table
a.基本转换
DataStream<Tuple2<String, Integer>> stream = ...;
Table table = tableEnv.fromDataStream(stream);
自动推断Schema
字段名f0, f1
b.指定字段名
Table table = tableEnv.fromDataStream(stream, $("name"), $("age"));
自定义字段名
c.指定时间属性
Table table = tableEnv.fromDataStream(
stream,
$("name"),
$("age"),
$("proc_time").proctime()
);
添加处理时间
d.事件时间
DataStream<Event> stream = env
.addSource(...)
.assignTimestampsAndWatermarks(...);
Table table = tableEnv.fromDataStream(
stream,
$("name"),
$("timestamp").rowtime()
);
指定事件时间
02.Table转DataStream
a.Append模式
DataStream<Row> stream = tableEnv.toDataStream(table);
只支持INSERT操作
适合无更新的查询
b.Retract模式
DataStream<Tuple2<Boolean, Row>> stream = tableEnv.toRetractStream(table, Row.class);
支持UPDATE和DELETE
Boolean表示添加(true)或撤回(false)
c.Upsert模式
tableEnv.toUpsertStream(table, Row.class);
支持UPDATE
需要有唯一键
d.Changelog模式
DataStream<Row> stream = tableEnv.toChangelogStream(table);
完整变更日志
包含操作类型
03.转换注意事项
a.模式选择
无聚合:Append模式
有聚合:Retract或Upsert模式
根据查询类型选择
b.类型映射
Flink类型与Table类型对应
注意类型转换
使用TypeInformation
c.性能考虑
频繁转换影响性能
尽量在Table API中完成操作
减少转换次数
4.8 动态表和持续查询
01.动态表概念
a.定义
流数据的表抽象
随时间变化的表
持续更新
b.流表对偶性
流可以转换为动态表
动态表可以转换为流
统一抽象
c.变更日志
INSERT:新增行
UPDATE:更新行
DELETE:删除行
表的变化记录
02.持续查询
a.定义
在动态表上的查询
持续执行
结果持续更新
b.查询类型
Append查询:只有INSERT
Update查询:有UPDATE
Retract查询:有DELETE
c.执行模式
流式执行
增量计算
实时结果
03.结果更新模式
a.Append模式
只输出新增行
适合无聚合查询
或窗口聚合
b.Retract模式
输出新增和撤回
适合有聚合的查询
完整语义
c.Upsert模式
输出插入和更新
需要唯一键
高效更新
04.状态管理
a.查询状态
聚合查询需要状态
Join查询需要状态
状态大小影响性能
b.状态清理
配置状态TTL
定期清理过期状态
避免状态膨胀
c.状态后端
选择合适的StateBackend
RocksDB适合大状态
内存适合小状态
5 Connector API使用
5.1 Kafka Connector
01.Kafka Source
a.基本配置
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "test-group");
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
"my-topic",
new SimpleStringSchema(),
properties
);
DataStream<String> stream = env.addSource(consumer);
创建Kafka消费者
指定topic和反序列化
b.多topic消费
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
Arrays.asList("topic1", "topic2", "topic3"),
new SimpleStringSchema(),
properties
);
消费多个topic
c.正则匹配topic
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
Pattern.compile("test-topic-[0-9]+"),
new SimpleStringSchema(),
properties
);
正则匹配topic名称
动态订阅
d.起始位置
consumer.setStartFromEarliest(); // 从最早位置
consumer.setStartFromLatest(); // 从最新位置
consumer.setStartFromTimestamp(1592323200000L); // 从时间戳
consumer.setStartFromGroupOffsets(); // 从group offset(默认)
指定消费起始位置
02.Kafka Sink
a.基本配置
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(
"my-topic",
new SimpleStringSchema(),
properties
);
stream.addSink(producer);
创建Kafka生产者
写入topic
b.自定义分区
FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(
"my-topic",
new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()),
properties,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE
);
producer.setWriteTimestampToKafka(true);
自定义分区策略
写入时间戳
c.语义保证
FlinkKafkaProducer.Semantic.NONE // 无保证
FlinkKafkaProducer.Semantic.AT_LEAST_ONCE // 至少一次
FlinkKafkaProducer.Semantic.EXACTLY_ONCE // 精确一次
选择语义级别
03.序列化和反序列化
a.SimpleStringSchema
new SimpleStringSchema()
字符串序列化
简单场景
b.JSONKeyValueDeserializationSchema
new JSONKeyValueDeserializationSchema(false)
JSON反序列化
包含元数据
c.自定义Schema
public class MyDeserializationSchema implements DeserializationSchema<MyEvent> {
@Override
public MyEvent deserialize(byte[] message) {
// 自定义反序列化逻辑
return parseEvent(message);
}
@Override
public boolean isEndOfStream(MyEvent nextElement) {
return false;
}
@Override
public TypeInformation<MyEvent> getProducedType() {
return TypeInformation.of(MyEvent.class);
}
}
实现DeserializationSchema
自定义逻辑
04.Exactly-Once保证
a.Source端
启用Checkpoint
Kafka offset提交到Checkpoint
故障恢复从Checkpoint恢复
b.Sink端
使用Kafka事务
两阶段提交
Checkpoint完成时提交事务
c.配置要求
Kafka 0.11+版本
启用Checkpoint
设置EXACTLY_ONCE语义
配置transaction.timeout.ms
5.2 JDBC Connector
01.JDBC Source
a.基本配置
JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:mysql://localhost:3306/test")
.withDriverName("com.mysql.jdbc.Driver")
.withUsername("root")
.withPassword("password")
.build();
DataStream<MyPojo> stream = env.createInput(
JdbcInputFormat.buildJdbcInputFormat()
.setDrivername("com.mysql.jdbc.Driver")
.setDBUrl("jdbc:mysql://localhost:3306/test")
.setUsername("root")
.setPassword("password")
.setQuery("SELECT * FROM my_table")
.setRowTypeInfo(TypeInformation.of(MyPojo.class))
.finish()
);
JDBC读取
批量读取
b.分区读取
JdbcInputFormat.buildJdbcInputFormat()
.setQuery("SELECT * FROM my_table WHERE id >= ? AND id < ?")
.setParametersProvider(new JdbcParameterValuesProvider() {
@Override
public Serializable[][] getParameterValues() {
return new Long[][]{
{0L, 1000L},
{1000L, 2000L},
{2000L, 3000L}
};
}
})
.finish();
并行读取
分区参数
02.JDBC Sink
a.基本配置
stream.addSink(
JdbcSink.sink(
"INSERT INTO my_table (id, name, age) VALUES (?, ?, ?)",
(statement, value) -> {
statement.setLong(1, value.getId());
statement.setString(2, value.getName());
statement.setInt(3, value.getAge());
},
JdbcExecutionOptions.builder()
.withBatchSize(1000)
.withBatchIntervalMs(200)
.withMaxRetries(5)
.build(),
connectionOptions
)
);
JDBC写入
批量写入
b.执行选项
withBatchSize:批次大小
withBatchIntervalMs:批次间隔
withMaxRetries:最大重试次数
优化性能
c.Upsert模式
JdbcSink.sink(
"INSERT INTO my_table (id, name) VALUES (?, ?) " +
"ON DUPLICATE KEY UPDATE name = VALUES(name)",
(statement, value) -> {
statement.setLong(1, value.getId());
statement.setString(2, value.getName());
},
executionOptions,
connectionOptions
);
插入或更新
幂等写入
03.Table API JDBC
a.创建JDBC表
tableEnv.executeSql(
"CREATE TABLE jdbc_table (" +
" id BIGINT," +
" name STRING," +
" age INT," +
" PRIMARY KEY (id) NOT ENFORCED" +
") WITH (" +
" 'connector' = 'jdbc'," +
" 'url' = 'jdbc:mysql://localhost:3306/test'," +
" 'table-name' = 'my_table'," +
" 'username' = 'root'," +
" 'password' = 'password'" +
")"
);
DDL创建JDBC表
Table API使用
b.读取JDBC表
Table table = tableEnv.from("jdbc_table");
SQL查询JDBC表
c.写入JDBC表
tableEnv.executeSql(
"INSERT INTO jdbc_table SELECT * FROM source_table"
);
SQL写入JDBC表
04.连接池配置
a.HikariCP配置
添加HikariCP依赖
配置连接池参数
提高性能
b.连接复用
批量操作
减少连接创建
提高吞吐量
5.3 Elasticsearch Connector
01.Elasticsearch Sink
a.基本配置
List<HttpHost> httpHosts = new ArrayList<>();
httpHosts.add(new HttpHost("localhost", 9200, "http"));
ElasticsearchSink.Builder<MyEvent> esSinkBuilder = new ElasticsearchSink.Builder<>(
httpHosts,
new ElasticsearchSinkFunction<MyEvent>() {
@Override
public void process(MyEvent element, RuntimeContext ctx, RequestIndexer indexer) {
Map<String, Object> json = new HashMap<>();
json.put("id", element.getId());
json.put("name", element.getName());
IndexRequest request = Requests.indexRequest()
.index("my-index")
.id(element.getId().toString())
.source(json);
indexer.add(request);
}
}
);
stream.addSink(esSinkBuilder.build());
写入Elasticsearch
索引文档
b.批量配置
esSinkBuilder.setBulkFlushMaxActions(1000);
esSinkBuilder.setBulkFlushMaxSizeMb(5);
esSinkBuilder.setBulkFlushInterval(5000);
批量写入配置
优化性能
c.失败处理
esSinkBuilder.setFailureHandler(new ActionRequestFailureHandler() {
@Override
public void onFailure(ActionRequest action, Throwable failure,
int restStatusCode, RequestIndexer indexer) {
// 处理失败
if (restStatusCode == 429) {
// 重试
indexer.add(action);
}
}
});
自定义失败处理
重试策略
02.Table API Elasticsearch
a.创建ES表
tableEnv.executeSql(
"CREATE TABLE es_table (" +
" id BIGINT," +
" name STRING," +
" age INT" +
") WITH (" +
" 'connector' = 'elasticsearch-7'," +
" 'hosts' = 'http://localhost:9200'," +
" 'index' = 'my-index'" +
")"
);
DDL创建ES表
b.写入ES
tableEnv.executeSql(
"INSERT INTO es_table SELECT * FROM source_table"
);
SQL写入ES
5.4 HBase Connector
01.HBase Sink
a.基本配置
Configuration hbaseConf = HBaseConfiguration.create();
hbaseConf.set(HConstants.ZOOKEEPER_QUORUM, "localhost:2181");
stream.addSink(new HBaseSinkFunction<MyEvent>(
"my_table",
hbaseConf,
new HBaseMutationConverter<MyEvent>() {
@Override
public void open() {}
@Override
public Mutation convertToMutation(MyEvent record) {
Put put = new Put(Bytes.toBytes(record.getId()));
put.addColumn(
Bytes.toBytes("cf"),
Bytes.toBytes("name"),
Bytes.toBytes(record.getName())
);
return put;
}
}
));
写入HBase
Put操作
b.批量写入
配置批量大小
缓冲区设置
提高性能
02.Table API HBase
a.创建HBase表
tableEnv.executeSql(
"CREATE TABLE hbase_table (" +
" rowkey STRING," +
" cf ROW<name STRING, age INT>," +
" PRIMARY KEY (rowkey) NOT ENFORCED" +
") WITH (" +
" 'connector' = 'hbase-2.2'," +
" 'table-name' = 'my_table'," +
" 'zookeeper.quorum' = 'localhost:2181'" +
")"
);
DDL创建HBase表
列族定义
b.读写HBase
读取:SELECT * FROM hbase_table
写入:INSERT INTO hbase_table SELECT * FROM source_table
SQL操作HBase
5.5 File System Connector
01.读取文件
a.文本文件
DataStream<String> stream = env.readTextFile("hdfs://path/to/file");
读取文本文件
HDFS或本地
b.CSV文件
CsvReaderFormat<MyPojo> format = CsvReaderFormat.forPojo(MyPojo.class);
FileSource<MyPojo> source = FileSource
.forRecordStreamFormat(format, new Path("hdfs://path"))
.build();
env.fromSource(source, WatermarkStrategy.noWatermarks(), "file-source");
读取CSV文件
POJO映射
c.Parquet文件
AvroParquetReaders.forSpecificRecord(MyAvroClass.class);
读取Parquet文件
Avro格式
02.写入文件
a.文本文件
stream.writeAsText("hdfs://path/to/output");
写入文本文件
b.StreamingFileSink
StreamingFileSink<String> sink = StreamingFileSink
.forRowFormat(
new Path("hdfs://path/to/output"),
new SimpleStringEncoder<String>("UTF-8")
)
.withRollingPolicy(
DefaultRollingPolicy.builder()
.withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
.withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
.withMaxPartSize(1024 * 1024 * 1024)
.build()
)
.build();
stream.addSink(sink);
流式文件Sink
滚动策略
Exactly-Once保证
c.Parquet格式
StreamingFileSink<MyAvroClass> sink = StreamingFileSink
.forBulkFormat(
new Path("hdfs://path/to/output"),
ParquetAvroWriters.forSpecificRecord(MyAvroClass.class)
)
.build();
写入Parquet文件
列式存储
03.Table API文件
a.创建文件表
tableEnv.executeSql(
"CREATE TABLE file_table (" +
" id BIGINT," +
" name STRING," +
" age INT" +
") WITH (" +
" 'connector' = 'filesystem'," +
" 'path' = 'hdfs://path/to/data'," +
" 'format' = 'csv'" +
")"
);
DDL创建文件表
支持多种格式
b.分区表
tableEnv.executeSql(
"CREATE TABLE partitioned_table (" +
" id BIGINT," +
" name STRING," +
" dt STRING," +
" hr STRING" +
") PARTITIONED BY (dt, hr) " +
"WITH (" +
" 'connector' = 'filesystem'," +
" 'path' = 'hdfs://path/to/data'," +
" 'format' = 'parquet'" +
")"
);
分区表定义
动态分区写入
5.6 自定义Source开发
01.SourceFunction接口
a.基本实现
public class MySourceFunction implements SourceFunction<MyEvent> {
private volatile boolean isRunning = true;
@Override
public void run(SourceContext<MyEvent> ctx) throws Exception {
while (isRunning) {
MyEvent event = generateEvent();
ctx.collect(event);
Thread.sleep(100);
}
}
@Override
public void cancel() {
isRunning = false;
}
}
实现SourceFunction
run方法生成数据
cancel方法停止
b.使用自定义Source
DataStream<MyEvent> stream = env.addSource(new MySourceFunction());
添加到环境
02.RichSourceFunction
a.生命周期方法
public class MyRichSource extends RichSourceFunction<MyEvent> {
private transient Connection connection;
@Override
public void open(Configuration parameters) {
connection = createConnection();
}
@Override
public void run(SourceContext<MyEvent> ctx) {
// 使用connection读取数据
}
@Override
public void close() {
if (connection != null) {
connection.close();
}
}
@Override
public void cancel() {
isRunning = false;
}
}
open初始化资源
close清理资源
访问RuntimeContext
03.并行Source
a.ParallelSourceFunction
public class MyParallelSource implements ParallelSourceFunction<MyEvent> {
@Override
public void run(SourceContext<MyEvent> ctx) {
int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
// 根据子任务索引读取不同分区
}
@Override
public void cancel() {
isRunning = false;
}
}
支持并行
多个实例
b.分区读取
根据子任务索引
读取不同数据分区
提高吞吐量
04.Checkpoint支持
a.CheckpointedFunction
public class MyCheckpointedSource extends RichSourceFunction<MyEvent>
implements CheckpointedFunction {
private ListState<Long> offsetState;
private Long offset = 0L;
@Override
public void initializeState(FunctionInitializationContext context) {
offsetState = context.getOperatorStateStore()
.getListState(new ListStateDescriptor<>("offset", Long.class));
for (Long o : offsetState.get()) {
offset = o;
}
}
@Override
public void snapshotState(FunctionSnapshotContext context) {
offsetState.clear();
offsetState.add(offset);
}
@Override
public void run(SourceContext<MyEvent> ctx) {
while (isRunning) {
MyEvent event = readFromOffset(offset);
synchronized (ctx.getCheckpointLock()) {
ctx.collect(event);
offset++;
}
}
}
}
实现CheckpointedFunction
保存和恢复状态
Exactly-Once保证
5.7 自定义Sink开发
01.SinkFunction接口
a.基本实现
public class MySinkFunction implements SinkFunction<MyEvent> {
@Override
public void invoke(MyEvent value, Context context) {
// 写入外部系统
writeToExternalSystem(value);
}
}
实现SinkFunction
invoke方法处理每个元素
b.使用自定义Sink
stream.addSink(new MySinkFunction());
添加到流
02.RichSinkFunction
a.生命周期方法
public class MyRichSink extends RichSinkFunction<MyEvent> {
private transient Connection connection;
@Override
public void open(Configuration parameters) {
connection = createConnection();
}
@Override
public void invoke(MyEvent value, Context context) {
connection.write(value);
}
@Override
public void close() {
if (connection != null) {
connection.close();
}
}
}
open初始化连接
close清理资源
复用连接
03.批量Sink
a.缓冲批量写入
public class BatchSink extends RichSinkFunction<MyEvent> {
private List<MyEvent> buffer = new ArrayList<>();
private static final int BATCH_SIZE = 1000;
@Override
public void invoke(MyEvent value, Context context) {
buffer.add(value);
if (buffer.size() >= BATCH_SIZE) {
flush();
}
}
private void flush() {
if (!buffer.isEmpty()) {
batchWrite(buffer);
buffer.clear();
}
}
@Override
public void close() {
flush();
}
}
缓冲批量写入
提高性能
close时刷新
04.TwoPhaseCommitSinkFunction
a.Exactly-Once Sink
public class MyTwoPhaseCommitSink extends TwoPhaseCommitSinkFunction<
MyEvent,
MyTransaction,
Void> {
@Override
protected MyTransaction beginTransaction() {
return new MyTransaction();
}
@Override
protected void invoke(MyTransaction transaction, MyEvent value, Context context) {
transaction.write(value);
}
@Override
protected void preCommit(MyTransaction transaction) {
transaction.preCommit();
}
@Override
protected void commit(MyTransaction transaction) {
transaction.commit();
}
@Override
protected void abort(MyTransaction transaction) {
transaction.rollback();
}
}
两阶段提交
Exactly-Once保证
事务支持
b.应用场景
数据库写入
文件系统写入
需要事务保证的场景
6 状态管理API
6.1 Keyed State
01.ValueState
a.定义和使用
public class MyMapFunction extends RichMapFunction<Event, Result> {
private transient ValueState<Long> countState;
@Override
public void open(Configuration parameters) {
ValueStateDescriptor<Long> descriptor =
new ValueStateDescriptor<>("count", Long.class);
countState = getRuntimeContext().getState(descriptor);
}
@Override
public Result map(Event value) throws Exception {
Long count = countState.value();
if (count == null) {
count = 0L;
}
count++;
countState.update(count);
return new Result(value.getKey(), count);
}
}
单个值状态
每个key一个值
读取和更新
b.API方法
value():获取当前值
update(T value):更新值
clear():清除状态
简单状态管理
02.ListState
a.定义和使用
public class MyProcessFunction extends KeyedProcessFunction<String, Event, Result> {
private transient ListState<Event> eventListState;
@Override
public void open(Configuration parameters) {
ListStateDescriptor<Event> descriptor =
new ListStateDescriptor<>("events", Event.class);
eventListState = getRuntimeContext().getListState(descriptor);
}
@Override
public void processElement(Event value, Context ctx, Collector<Result> out)
throws Exception {
eventListState.add(value);
List<Event> events = new ArrayList<>();
for (Event e : eventListState.get()) {
events.add(e);
}
if (events.size() >= 10) {
out.collect(new Result(events));
eventListState.clear();
}
}
}
列表状态
存储多个值
遍历和添加
b.API方法
add(T value):添加元素
get():获取所有元素
addAll(List<T> values):批量添加
update(List<T> values):更新列表
clear():清除状态
03.MapState
a.定义和使用
public class MyProcessFunction extends KeyedProcessFunction<String, Event, Result> {
private transient MapState<String, Long> mapState;
@Override
public void open(Configuration parameters) {
MapStateDescriptor<String, Long> descriptor =
new MapStateDescriptor<>("map", String.class, Long.class);
mapState = getRuntimeContext().getMapState(descriptor);
}
@Override
public void processElement(Event value, Context ctx, Collector<Result> out)
throws Exception {
String key = value.getSubKey();
Long count = mapState.get(key);
if (count == null) {
count = 0L;
}
count++;
mapState.put(key, count);
out.collect(new Result(value.getKey(), mapState.entries()));
}
}
Map状态
键值对存储
灵活查询
b.API方法
get(UK key):获取值
put(UK key, UV value):存储键值对
contains(UK key):检查是否存在
remove(UK key):删除键
entries():获取所有条目
keys():获取所有键
values():获取所有值
isEmpty():检查是否为空
clear():清除状态
04.ReducingState
a.定义和使用
public class MyProcessFunction extends KeyedProcessFunction<String, Event, Result> {
private transient ReducingState<Long> reducingState;
@Override
public void open(Configuration parameters) {
ReducingStateDescriptor<Long> descriptor =
new ReducingStateDescriptor<>(
"sum",
new ReduceFunction<Long>() {
@Override
public Long reduce(Long value1, Long value2) {
return value1 + value2;
}
},
Long.class
);
reducingState = getRuntimeContext().getReducingState(descriptor);
}
@Override
public void processElement(Event value, Context ctx, Collector<Result> out)
throws Exception {
reducingState.add(value.getValue());
Long sum = reducingState.get();
out.collect(new Result(value.getKey(), sum));
}
}
归约状态
自动聚合
高效累加
b.API方法
add(T value):添加值并归约
get():获取归约结果
clear():清除状态
05.AggregatingState
a.定义和使用
public class MyProcessFunction extends KeyedProcessFunction<String, Event, Result> {
private transient AggregatingState<Long, Double> aggState;
@Override
public void open(Configuration parameters) {
AggregatingStateDescriptor<Long, Tuple2<Long, Long>, Double> descriptor =
new AggregatingStateDescriptor<>(
"average",
new AggregateFunction<Long, Tuple2<Long, Long>, Double>() {
@Override
public Tuple2<Long, Long> createAccumulator() {
return Tuple2.of(0L, 0L);
}
@Override
public Tuple2<Long, Long> add(Long value, Tuple2<Long, Long> acc) {
return Tuple2.of(acc.f0 + value, acc.f1 + 1);
}
@Override
public Double getResult(Tuple2<Long, Long> acc) {
return acc.f1 == 0 ? 0.0 : (double) acc.f0 / acc.f1;
}
@Override
public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
}
},
Types.TUPLE(Types.LONG, Types.LONG)
);
aggState = getRuntimeContext().getAggregatingState(descriptor);
}
@Override
public void processElement(Event value, Context ctx, Collector<Result> out)
throws Exception {
aggState.add(value.getValue());
Double avg = aggState.get();
out.collect(new Result(value.getKey(), avg));
}
}
聚合状态
自定义聚合逻辑
灵活累加器
b.API方法
add(IN value):添加值并聚合
get():获取聚合结果
clear():清除状态
6.2 Operator State
01.ListState
a.定义和使用
public class MySource extends RichParallelSourceFunction<Event>
implements CheckpointedFunction {
private transient ListState<Long> offsetState;
private Long offset = 0L;
@Override
public void initializeState(FunctionInitializationContext context) {
ListStateDescriptor<Long> descriptor =
new ListStateDescriptor<>("offset", Long.class);
offsetState = context.getOperatorStateStore().getListState(descriptor);
if (context.isRestored()) {
for (Long o : offsetState.get()) {
offset = Math.max(offset, o);
}
}
}
@Override
public void snapshotState(FunctionSnapshotContext context) {
offsetState.clear();
offsetState.add(offset);
}
@Override
public void run(SourceContext<Event> ctx) {
while (isRunning) {
Event event = readFromOffset(offset);
ctx.collect(event);
offset++;
}
}
}
Operator级别状态
所有并行实例共享
列表状态
b.状态重分布
Even-split:平均分配
Union:所有实例获取完整状态
根据场景选择
02.UnionListState
a.定义和使用
@Override
public void initializeState(FunctionInitializationContext context) {
ListStateDescriptor<String> descriptor =
new ListStateDescriptor<>("config", String.class);
configState = context.getOperatorStateStore().getUnionListState(descriptor);
if (context.isRestored()) {
for (String config : configState.get()) {
loadConfig(config);
}
}
}
Union重分布
所有实例获取完整状态
适合配置数据
b.应用场景
配置信息
规则数据
需要全量数据的场景
03.BroadcastState
a.定义和使用
MapStateDescriptor<String, Rule> ruleDescriptor =
new MapStateDescriptor<>("rules", String.class, Rule.class);
BroadcastStream<Rule> ruleBroadcast = ruleStream.broadcast(ruleDescriptor);
DataStream<Result> result = eventStream
.connect(ruleBroadcast)
.process(new KeyedBroadcastProcessFunction<String, Event, Rule, Result>() {
@Override
public void processElement(Event event, ReadOnlyContext ctx, Collector<Result> out)
throws Exception {
ReadOnlyBroadcastState<String, Rule> state =
ctx.getBroadcastState(ruleDescriptor);
Rule rule = state.get(event.getRuleId());
if (rule != null) {
out.collect(applyRule(event, rule));
}
}
@Override
public void processBroadcastElement(Rule rule, Context ctx, Collector<Result> out)
throws Exception {
BroadcastState<String, Rule> state = ctx.getBroadcastState(ruleDescriptor);
state.put(rule.getId(), rule);
}
});
广播状态
动态规则
所有实例共享
b.应用场景
动态配置
规则引擎
维度表Join
6.3 State TTL配置
01.启用State TTL
a.基本配置
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.hours(1))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build();
ValueStateDescriptor<Long> descriptor =
new ValueStateDescriptor<>("count", Long.class);
descriptor.enableTimeToLive(ttlConfig);
ValueState<Long> state = getRuntimeContext().getState(descriptor);
配置TTL
自动过期
节省内存
b.更新类型
OnCreateAndWrite:创建和写入时更新
OnReadAndWrite:读取和写入时更新
Disabled:不更新
控制TTL更新时机
c.可见性
NeverReturnExpired:永不返回过期值
ReturnExpiredIfNotCleanedUp:清理前可返回
控制过期值可见性
02.清理策略
a.全量快照清理
ttlConfig.cleanupFullSnapshot()
Checkpoint时清理
完整扫描
适合小状态
b.增量清理
ttlConfig.cleanupIncrementally(10, true)
每次访问清理部分
参数:每次清理条目数
适合大状态
c.RocksDB压缩清理
ttlConfig.cleanupInRocksdbCompactFilter(1000)
RocksDB压缩时清理
后台清理
适合RocksDB后端
03.TTL最佳实践
a.合理设置TTL
根据业务需求
避免过短或过长
监控状态大小
b.选择清理策略
小状态用全量清理
大状态用增量清理
RocksDB用压缩清理
c.性能考虑
TTL有性能开销
清理策略影响性能
权衡内存和性能
6.4 Queryable State
01.启用Queryable State
a.配置
queryable-state.enable:true
queryable-state.proxy.ports:9069
queryable-state.server.ports:9067
启用可查询状态
配置端口
b.声明可查询状态
ValueStateDescriptor<Long> descriptor =
new ValueStateDescriptor<>("count", Long.class);
descriptor.setQueryable("query-name");
ValueState<Long> state = getRuntimeContext().getState(descriptor);
设置查询名称
外部可查询
02.查询状态
a.创建客户端
QueryableStateClient client = new QueryableStateClient(
"localhost",
9069
);
连接到Queryable State Proxy
b.查询状态
CompletableFuture<ValueState<Long>> future = client.getKvState(
jobId,
"query-name",
key,
BasicTypeInfo.STRING_TYPE_INFO,
new ValueStateDescriptor<>("count", Long.class)
);
future.thenAccept(state -> {
Long value = state.value();
System.out.println("Count: " + value);
});
异步查询
获取状态值
03.应用场景
a.实时查询
外部系统查询Flink状态
实时获取计算结果
无需等待输出
b.监控和调试
查看状态值
调试应用
监控状态
c.注意事项
性能影响
网络开销
谨慎使用
6.5 State Backend选择
01.MemoryStateBackend
a.特点
状态存储在TaskManager内存
Checkpoint存储在JobManager内存
适合小状态
快速但不可靠
b.配置
env.setStateBackend(new MemoryStateBackend());
默认后端
开发测试使用
c.限制
状态大小受限于内存
JobManager内存限制Checkpoint大小
不适合生产环境
02.FsStateBackend
a.特点
状态存储在TaskManager内存
Checkpoint存储在文件系统
适合中等状态
平衡性能和可靠性
b.配置
env.setStateBackend(new FsStateBackend("hdfs://namenode:port/flink/checkpoints"));
指定文件系统路径
HDFS或本地文件系统
c.适用场景
中等大小状态
需要持久化
内存足够
03.RocksDBStateBackend
a.特点
状态存储在RocksDB(磁盘)
Checkpoint存储在文件系统
适合大状态
支持增量Checkpoint
b.配置
env.setStateBackend(new RocksDBStateBackend("hdfs://namenode:port/flink/checkpoints"));
env.getCheckpointConfig().enableIncrementalCheckpointing(true);
启用RocksDB
增量Checkpoint
c.优化配置
state.backend.rocksdb.predefined-options:SPINNING_DISK_OPTIMIZED
state.backend.rocksdb.block.cache-size:256m
state.backend.rocksdb.write-buffer-size:64m
调优参数
提高性能
d.适用场景
大状态应用
状态超过内存
需要增量Checkpoint
04.选择建议
a.开发测试
MemoryStateBackend
快速迭代
简单场景
b.中等状态
FsStateBackend
平衡性能
可靠性好
c.大状态
RocksDBStateBackend
支持大状态
增量Checkpoint
生产推荐
6.6 Checkpoint和Savepoint
01.Checkpoint配置
a.启用Checkpoint
env.enableCheckpointing(60000); // 60秒
设置Checkpoint间隔
自动容错
b.Checkpoint模式
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
EXACTLY_ONCE:精确一次
AT_LEAST_ONCE:至少一次
选择语义
c.Checkpoint配置
CheckpointConfig config = env.getCheckpointConfig();
config.setMinPauseBetweenCheckpoints(500); // 最小间隔
config.setCheckpointTimeout(60000); // 超时时间
config.setMaxConcurrentCheckpoints(1); // 最大并发
config.enableExternalizedCheckpoints(
CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);
详细配置
优化性能
02.Savepoint操作
a.触发Savepoint
bin/flink savepoint <jobId> [targetDirectory]
手动触发
指定保存路径
b.从Savepoint恢复
bin/flink run -s <savepointPath> <jobJar>
指定Savepoint路径
恢复作业
c.应用场景
版本升级
集群迁移
参数调整
手动备份
03.Checkpoint vs Savepoint
a.Checkpoint
自动触发
轻量级
用于故障恢复
可以自动清理
b.Savepoint
手动触发
完整快照
用于版本升级
需要手动管理
c.选择
日常容错用Checkpoint
升级迁移用Savepoint
结合使用
04.状态恢复
a.自动恢复
作业失败自动从最近Checkpoint恢复
无需人工干预
保证状态一致性
b.手动恢复
从Savepoint恢复
指定恢复点
灵活控制
c.状态兼容性
升级时注意状态兼容
使用UID标识算子
避免状态丢失
05.最佳实践
a.Checkpoint间隔
根据数据量和延迟要求
一般1-5分钟
平衡性能和恢复时间
b.状态大小
监控状态大小
及时清理
使用TTL
避免状态膨胀
c.Checkpoint存储
使用可靠存储
HDFS或S3
定期清理旧Checkpoint
保留足够空间
d.增量Checkpoint
大状态使用增量Checkpoint
减少Checkpoint时间
提高性能
RocksDB支持
7 高级API特性
7.1 ProcessFunction详解
01.ProcessFunction基础
a.基本用法
stream.process(new ProcessFunction<Event, Result>() {
@Override
public void processElement(Event value, Context ctx, Collector<Result> out)
throws Exception {
out.collect(new Result(value));
// 访问时间戳
Long timestamp = ctx.timestamp();
// 访问水位线
Long watermark = ctx.timerService().currentWatermark();
}
});
底层API
完全控制
访问上下文信息
b.Context功能
timestamp():元素时间戳
timerService():定时器服务
output(OutputTag, value):侧输出
getCurrentKey():当前键(KeyedStream)
丰富的上下文信息
02.KeyedProcessFunction
a.基本用法
keyedStream.process(new KeyedProcessFunction<String, Event, Result>() {
private ValueState<Long> countState;
@Override
public void open(Configuration parameters) {
ValueStateDescriptor<Long> descriptor =
new ValueStateDescriptor<>("count", Long.class);
countState = getRuntimeContext().getState(descriptor);
}
@Override
public void processElement(Event value, Context ctx, Collector<Result> out)
throws Exception {
Long count = countState.value();
if (count == null) {
count = 0L;
}
count++;
countState.update(count);
// 注册定时器
ctx.timerService().registerEventTimeTimer(value.getTimestamp() + 60000);
out.collect(new Result(ctx.getCurrentKey(), count));
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Result> out)
throws Exception {
out.collect(new Result(ctx.getCurrentKey(), "Timer triggered"));
}
});
Keyed流处理
访问状态
注册定时器
b.应用场景
复杂状态逻辑
定时器需求
侧输出
完全控制处理逻辑
03.ProcessWindowFunction
a.基本用法
keyedStream
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.process(new ProcessWindowFunction<Event, Result, String, TimeWindow>() {
@Override
public void process(
String key,
Context context,
Iterable<Event> elements,
Collector<Result> out) throws Exception {
long count = 0;
for (Event e : elements) {
count++;
}
TimeWindow window = context.window();
out.collect(new Result(
key,
count,
window.getStart(),
window.getEnd()
));
}
});
窗口全量处理
访问所有元素
访问窗口元数据
b.Context功能
window():窗口信息
currentProcessingTime():处理时间
currentWatermark():当前水位线
windowState():窗口状态
globalState():全局状态
output():侧输出
c.增量聚合+全量处理
keyedStream
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.aggregate(
new AverageAggregate(),
new ProcessWindowFunction<Double, Result, String, TimeWindow>() {
@Override
public void process(String key, Context context,
Iterable<Double> elements,
Collector<Result> out) {
Double avg = elements.iterator().next();
out.collect(new Result(key, avg, context.window()));
}
}
);
先增量聚合
再全量处理
兼顾性能和灵活性
04.CoProcessFunction
a.基本用法
connectedStream.process(new CoProcessFunction<Event1, Event2, Result>() {
private ValueState<Event1> event1State;
private ValueState<Event2> event2State;
@Override
public void open(Configuration parameters) {
event1State = getRuntimeContext().getState(
new ValueStateDescriptor<>("event1", Event1.class));
event2State = getRuntimeContext().getState(
new ValueStateDescriptor<>("event2", Event2.class));
}
@Override
public void processElement1(Event1 value, Context ctx, Collector<Result> out)
throws Exception {
event1State.update(value);
Event2 event2 = event2State.value();
if (event2 != null) {
out.collect(new Result(value, event2));
}
}
@Override
public void processElement2(Event2 value, Context ctx, Collector<Result> out)
throws Exception {
event2State.update(value);
Event1 event1 = event1State.value();
if (event1 != null) {
out.collect(new Result(event1, value));
}
}
});
处理连接流
共享状态
协同处理
b.应用场景
流关联
规则匹配
配置更新
双流Join
7.2 定时器使用
01.Processing Time Timer
a.注册定时器
@Override
public void processElement(Event value, Context ctx, Collector<Result> out)
throws Exception {
long currentTime = ctx.timerService().currentProcessingTime();
ctx.timerService().registerProcessingTimeTimer(currentTime + 60000);
}
注册处理时间定时器
60秒后触发
b.定时器回调
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Result> out)
throws Exception {
out.collect(new Result("Timer fired at: " + timestamp));
}
定时器触发时调用
处理超时逻辑
c.应用场景
超时检测
定期清理
心跳检测
简单定时任务
02.Event Time Timer
a.注册定时器
@Override
public void processElement(Event value, Context ctx, Collector<Result> out)
throws Exception {
ctx.timerService().registerEventTimeTimer(value.getTimestamp() + 60000);
}
注册事件时间定时器
基于事件时间
b.水位线触发
水位线超过定时器时间时触发
精确时间语义
处理乱序
c.应用场景
会话超时
窗口触发
延迟数据处理
精确时间控制
03.定时器管理
a.删除定时器
ctx.timerService().deleteProcessingTimeTimer(timestamp);
ctx.timerService().deleteEventTimeTimer(timestamp);
删除已注册的定时器
避免不必要触发
b.定时器状态
定时器自动持久化
Checkpoint包含定时器
故障恢复后继续
c.定时器去重
同一时间戳只注册一次
自动去重
避免重复触发
04.定时器最佳实践
a.合理使用
避免过多定时器
影响性能
及时删除
b.时间选择
处理时间:简单快速
事件时间:精确可靠
根据需求选择
c.状态清理
定时器触发时清理状态
避免状态膨胀
使用TTL辅助
7.3 Broadcast State Pattern
01.广播状态定义
a.创建广播流
MapStateDescriptor<String, Rule> ruleDescriptor =
new MapStateDescriptor<>("rules", String.class, Rule.class);
BroadcastStream<Rule> ruleBroadcast = ruleStream.broadcast(ruleDescriptor);
定义广播状态描述符
创建广播流
b.连接流
DataStream<Event> result = eventStream
.connect(ruleBroadcast)
.process(new BroadcastProcessFunction<Event, Rule, Result>() {
// 处理逻辑
});
连接数据流和广播流
协同处理
02.BroadcastProcessFunction
a.处理数据元素
@Override
public void processElement(Event event, ReadOnlyContext ctx, Collector<Result> out)
throws Exception {
ReadOnlyBroadcastState<String, Rule> state =
ctx.getBroadcastState(ruleDescriptor);
for (Map.Entry<String, Rule> entry : state.immutableEntries()) {
Rule rule = entry.getValue();
if (rule.matches(event)) {
out.collect(applyRule(event, rule));
}
}
}
读取广播状态
只读访问
应用规则
b.处理广播元素
@Override
public void processBroadcastElement(Rule rule, Context ctx, Collector<Result> out)
throws Exception {
BroadcastState<String, Rule> state = ctx.getBroadcastState(ruleDescriptor);
state.put(rule.getId(), rule);
}
更新广播状态
可写访问
动态更新规则
c.Context功能
getBroadcastState():获取广播状态
applyToKeyedState():访问Keyed状态
timestamp():时间戳
currentWatermark():水位线
03.KeyedBroadcastProcessFunction
a.Keyed流处理
keyedStream
.connect(ruleBroadcast)
.process(new KeyedBroadcastProcessFunction<String, Event, Rule, Result>() {
private ValueState<Long> countState;
@Override
public void open(Configuration parameters) {
countState = getRuntimeContext().getState(
new ValueStateDescriptor<>("count", Long.class));
}
@Override
public void processElement(Event event, ReadOnlyContext ctx,
Collector<Result> out) throws Exception {
Long count = countState.value();
if (count == null) count = 0L;
count++;
countState.update(count);
ReadOnlyBroadcastState<String, Rule> state =
ctx.getBroadcastState(ruleDescriptor);
Rule rule = state.get(event.getRuleId());
if (rule != null) {
out.collect(applyRule(event, rule, count));
}
}
@Override
public void processBroadcastElement(Rule rule, Context ctx,
Collector<Result> out) throws Exception {
ctx.getBroadcastState(ruleDescriptor).put(rule.getId(), rule);
// 可以访问所有Keyed状态
ctx.applyToKeyedState(
new ValueStateDescriptor<>("count", Long.class),
(key, state) -> {
// 处理每个key的状态
}
);
}
});
Keyed流+广播流
访问Keyed状态
更灵活
04.应用场景
a.动态规则
规则引擎
实时更新规则
无需重启作业
b.配置更新
动态配置
参数调整
实时生效
c.维度表Join
小表广播
实时更新维度
丰富流数据
7.4 CEP复杂事件处理
01.Pattern定义
a.简单模式
Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
.where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event event) {
return event.getType().equals("login");
}
})
.next("middle")
.where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event event) {
return event.getType().equals("pay");
}
})
.followedBy("end")
.where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event event) {
return event.getType().equals("logout");
}
})
.within(Time.minutes(10));
定义事件序列
严格连续:next
宽松连续:followedBy
时间窗口限制
b.量词模式
Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
.where(SimpleCondition.of(e -> e.getType().equals("error")))
.times(3) // 恰好3次
.within(Time.minutes(5));
times(n):恰好n次
times(n, m):n到m次
oneOrMore():一次或多次
timesOrMore(n):n次或更多
c.条件组合
pattern.where(condition1).or(condition2);
pattern.where(condition1).and(condition2);
逻辑组合
复杂条件
02.Pattern检测
a.PatternStream
PatternStream<Event> patternStream = CEP.pattern(
keyedStream,
pattern
);
应用模式到流
生成PatternStream
b.选择匹配
DataStream<Alert> result = patternStream.select(
new PatternSelectFunction<Event, Alert>() {
@Override
public Alert select(Map<String, List<Event>> pattern) {
Event start = pattern.get("start").get(0);
Event end = pattern.get("end").get(0);
return new Alert(start, end);
}
}
);
提取匹配事件
生成结果
c.flatSelect
DataStream<Alert> result = patternStream.flatSelect(
new PatternFlatSelectFunction<Event, Alert>() {
@Override
public void flatSelect(Map<String, List<Event>> pattern,
Collector<Alert> out) {
for (Event event : pattern.get("middle")) {
out.collect(new Alert(event));
}
}
}
);
扁平化输出
多个结果
03.超时处理
a.定义超时标签
OutputTag<TimeoutEvent> timeoutTag =
new OutputTag<TimeoutEvent>("timeout"){};
侧输出标签
b.处理超时
SingleOutputStreamOperator<Alert> result = patternStream.flatSelect(
timeoutTag,
new PatternFlatTimeoutFunction<Event, TimeoutEvent>() {
@Override
public void timeout(Map<String, List<Event>> pattern,
long timeoutTimestamp,
Collector<TimeoutEvent> out) {
out.collect(new TimeoutEvent(pattern, timeoutTimestamp));
}
},
new PatternFlatSelectFunction<Event, Alert>() {
@Override
public void flatSelect(Map<String, List<Event>> pattern,
Collector<Alert> out) {
out.collect(new Alert(pattern));
}
}
);
DataStream<TimeoutEvent> timeoutStream = result.getSideOutput(timeoutTag);
超时事件处理
侧输出流
04.CEP应用场景
a.欺诈检测
异常行为模式
多步骤欺诈
实时告警
b.用户行为分析
行为序列分析
转化漏斗
用户画像
c.设备监控
故障模式识别
预警告警
设备状态跟踪
d.业务流程监控
流程合规检查
SLA监控
异常流程识别
7.5 Gelly图计算API
01.图创建
a.从边创建
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<Edge<Long, Double>> edges = env.fromElements(
new Edge<>(1L, 2L, 0.5),
new Edge<>(2L, 3L, 0.6),
new Edge<>(3L, 4L, 0.7)
);
Graph<Long, NullValue, Double> graph = Graph.fromDataSet(edges, env);
从边集创建图
自动生成顶点
b.从顶点和边创建
DataSet<Vertex<Long, String>> vertices = env.fromElements(
new Vertex<>(1L, "A"),
new Vertex<>(2L, "B"),
new Vertex<>(3L, "C")
);
Graph<Long, String, Double> graph = Graph.fromDataSet(vertices, edges, env);
指定顶点和边
完整图定义
02.图操作
a.基本操作
graph.getVertices() // 获取顶点
graph.getEdges() // 获取边
graph.numberOfVertices() // 顶点数
graph.numberOfEdges() // 边数
graph.getDegrees() // 度数
基本图信息
b.转换操作
graph.mapVertices(v -> new Vertex<>(v.getId(), v.getValue().toUpperCase()))
graph.mapEdges(e -> new Edge<>(e.getSource(), e.getTarget(), e.getValue() * 2))
转换顶点和边
c.过滤操作
graph.filterOnVertices(v -> v.getValue().startsWith("A"))
graph.filterOnEdges(e -> e.getValue() > 0.5)
过滤图元素
03.图算法
a.PageRank
DataSet<Vertex<Long, Double>> result = graph.run(new PageRank<>(0.85, 10));
网页排名算法
迭代计算
b.连通分量
DataSet<Vertex<Long, Long>> result = graph.run(new ConnectedComponents<>(10));
查找连通分量
图分析
c.单源最短路径
DataSet<Vertex<Long, Double>> result = graph.run(
new SingleSourceShortestPaths<>(1L, 10)
);
最短路径算法
源顶点到所有顶点
d.三角形计数
DataSet<Tuple3<Long, Long, Long>> triangles = graph.run(new TriangleListing<>());
Long count = graph.run(new TriangleCount<>());
三角形枚举和计数
图密度分析
04.自定义图算法
a.Vertex-Centric迭代
graph.runVertexCentricIteration(
new VertexUpdateFunction<>(),
new MessagingFunction<>(),
maxIterations
);
顶点中心迭代
消息传递
b.Scatter-Gather迭代
graph.runScatterGatherIteration(
new ScatterFunction<>(),
new GatherFunction<>(),
maxIterations
);
分散-聚集模式
图计算框架
7.6 Python API(PyFlink)
01.环境配置
a.安装PyFlink
pip install apache-flink
安装Python包
b.创建环境
from pyflink.datastream import StreamExecutionEnvironment
env = StreamExecutionEnvironment.get_execution_environment()
创建执行环境
c.配置
env.set_parallelism(4)
env.enable_checkpointing(60000)
配置参数
02.DataStream API
a.创建流
ds = env.from_collection([1, 2, 3, 4, 5])
ds = env.from_source(...)
创建数据流
b.转换操作
ds.map(lambda x: x * 2)
ds.filter(lambda x: x > 0)
ds.flat_map(lambda x: [(x, x * 2)])
Python Lambda表达式
c.输出
ds.print()
ds.add_sink(...)
输出结果
03.Table API
a.创建表环境
from pyflink.table import StreamTableEnvironment
table_env = StreamTableEnvironment.create(env)
Table环境
b.SQL查询
table_env.execute_sql(
"CREATE TABLE source_table (...) WITH (...)"
)
result = table_env.sql_query("SELECT * FROM source_table")
SQL操作
c.Table API
from pyflink.table.expressions import col
result = table.select(col('name'), col('age')).where(col('age') > 18)
Table API操作
04.UDF定义
a.标量UDF
from pyflink.table.udf import udf
@udf(result_type=DataTypes.STRING())
def to_upper(s):
return s.upper()
table.select(to_upper(col('name')))
Python UDF
b.表UDF
from pyflink.table.udf import udtf
@udtf(result_types=[DataTypes.STRING(), DataTypes.INT()])
def split(s):
for word in s.split():
yield word, len(word)
Python UDTF
8 部署运维
8.1 部署模式
01.本地部署
a.单机模式
a.本地执行环境
LocalEnvironment：创建本地执行环境进行开发和测试
并行度设置：在本地环境中设置适当的并行度
调试工具：集成IDE调试工具进行断点调试
日志配置：配置本地日志输出格式和级别
示例代码：
---
// 创建本地执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
// 设置并行度
env.setParallelism(4);
// 启用检查点
env.enableCheckpointing(5000);
// 配置重启策略
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // 重启次数
Time.seconds(10) // 重启间隔
));
---
b.调试和开发
IDE集成：支持IntelliJ IDEA、Eclipse等主流IDE
断点调试：在UserFunction和Operator中设置断点
日志输出：本地日志实时输出和控制
参数传递：通过程序参数传递配置信息
测试框架：使用MiniCluster进行集成测试
示例代码：
---
// 使用MiniCluster进行测试
public class FlinkJobTest {
@Test
public void testFlinkJob() throws Exception {
Configuration config = new Configuration();
config.setString("execution.checkpointing.interval", "5s");
MiniClusterResource miniClusterResource = new MiniClusterResource(
new MiniClusterConfiguration.Builder()
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(4)
.setConfiguration(config)
.build()
);
miniClusterResource.before();
try {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 测试作业逻辑
TestSink<String> sink = new TestSink<>();
env.fromElements("test", "data")
.addSink(sink);
env.execute("Test Job");
// 验证结果
assertEquals(2, sink.getCollectedOutput().size());
} finally {
miniClusterResource.after();
}
}
}
---
c.IDE集成
Maven依赖：配置Flink核心和连接器依赖
源码下载：下载Flink源码便于调试
远程调试：配置远程调试端口
热部署：支持代码热部署和调试
项目模板：使用Flink项目快速启动模板
b.集群模式
a.Standalone集群
集群架构：JobManager和TaskManager分离部署
配置文件：flink-conf.yaml、masters、workers配置
启动脚本：start-cluster.sh、stop-cluster.sh管理脚本
集群管理：通过命令行工具管理集群状态
示例配置：
---yaml
# flink-conf.yaml
jobmanager.rpc.address: jobmanager-host
jobmanager.rpc.port: 6123
jobmanager.memory.process.size: 1600m
taskmanager.memory.process.size: 1728m
taskmanager.numberOfTaskSlots: 4
parallelism.default: 2
execution.checkpointing.interval: 5s
state.backend: rocksdb
state.checkpoints.dir: file:///checkpoint/dir
state.savepoints.dir: file:///savepoint/dir
---
b.资源配置
内存配置：JVM堆内存、托管内存、网络内存配置
CPU配置：TaskSlot数量和并行度配置
磁盘配置：检查点和状态存储配置
网络配置：网络缓冲区和通信配置
调优参数：GC策略、序列化配置等
c.高可用配置
ZooKeeper集成：使用ZooKeeper实现JobManager高可用
元数据存储：配置JobManager元数据存储后端
故障恢复：JobManager故障自动恢复机制
检查点恢复：从检查点自动恢复作业状态
示例配置：
---yaml
# 高可用配置
high-availability: zookeeper
high-availability.zookeeper.quorum: zk1:2181,zk2:2181,zk3:2181
high-availability.zookeeper.path.root: /flink
high-availability.cluster-id: /default
high-availability.storageDir: hdfs://namenode:port/flink/recovery
---
d.监控配置
Web UI：配置Web界面访问地址和端口
指标系统：配置指标报告器和监控系统
日志收集：配置日志收集和分析系统
告警配置：设置关键指标的告警规则
性能监控：集成APM工具进行性能监控
02.YARN部署
a.YARN Session模式
a.长期运行的YARN应用
Session集群：预先启动长期运行的Flink集群
资源申请：通过YARN申请固定的集群资源
作业提交：向Session集群提交多个作业
资源共享：多个作业共享集群资源
生命周期：集群独立于具体作业运行
b.资源预分配
Container配置：配置JobManager和TaskManager容器
内存分配：YARN容器内存和Flink进程内存配置
vCore分配：配置每个容器的虚拟CPU核心数
节点标签：使用YARN节点标签进行资源调度
队列配置：指定YARN资源队列和优先级
示例命令：
---bash
# 启动YARN Session集群
./bin/yarn-session.sh -d \
-n 4 \
-s 4 \
-jm 1024m \
-tm 4096m \
-nm "flink-session-cluster" \
-qu "default" \
--queue "default"
---
c.作业提交机制
提交方式：通过命令行或API向Session集群提交作业
端口映射：YARN端口映射和防火墙配置
依赖管理：作业依赖包的上传和管理
类路径：管理作业和集群的类路径冲突
动态加载：支持作业的动态加载和卸载
b.YARN Per-Job模式
a.每个作业一个YARN应用
作业隔离：每个作业运行在独立的YARN应用中
资源按需：根据作业需求动态申请资源
生命周期：资源随作业结束自动释放
故障隔离：作业故障不影响其他作业
成本控制：精确控制每个作业的资源使用
b.资源动态分配
自动扩缩：根据作业负载动态调整资源
弹性调度：YARN的弹性资源调度能力
资源抢占：高优先级作业的资源抢占机制
按需付费：云环境中的按需资源付费模式
成本优化：通过资源优化降低使用成本
示例命令：
---bash
# 提交Per-Job作业
./bin/flink run -m yarn-cluster \
-yjm 1024m \
-ytm 4096m \
-yn 4 \
-ys 4 \
-yqu "default" \
-c com.example.MyJob \
./job.jar \
--input /path/to/input \
--output /path/to/output
---
c.作业隔离
进程隔离：每个作业运行在独立的进程中
类加载隔离：作业间的类加载器隔离
网络隔离：作业间的网络通信隔离
安全隔离：作业的安全权限和访问控制
故障隔离：单作业故障不影响集群稳定性
c.YARN Application模式
a.Application Master执行作业
应用模式：JobManager运行在Application Master中
本地提交：作业从客户端本地提交
资源整合：更紧密的YARN资源整合
生命周期管理：应用生命周期管理更加精细
运维简化：简化了集群运维和管理
b.资源利用率优化
资源共享：Application Master和TaskManager资源共享
弹性调整：根据作业需求弹性调整资源
梯度调度：支持梯度资源调度策略
抢占机制：资源优先级和抢占机制
回收策略：资源回收和再利用策略
示例配置：
---bash
# Application模式提交作业
./bin/flink run-application \
-t yarn-application \
-Dyarn.application.name="my-flink-app" \
-Dyarn.application.queue="default" \
-Dtaskmanager.memory.process.size=4096m \
-Dtaskmanager.numberOfTaskSlots=4 \
-Dparallelism.default=2 \
-c com.example.MyJob \
./job.jar
---
c.故障隔离
应用级别：Application级别的故障隔离
进程管理：Application Master进程管理
健康检查：YARN健康检查机制
故障恢复：Application故障恢复策略
日志收集：YARN应用日志收集和分析
03.Kubernetes部署
a.K8s Session模式
a.Flink集群作为K8s服务
Service对象：创建Kubernetes Service暴露服务
Pod配置：JobManager和TaskManager Pod配置
配置管理：使用ConfigMap管理配置文件
存储管理：使用PersistentVolume管理持久化存储
网络配置：Kubernetes网络策略和服务发现
b.资源配置和管理
资源请求：CPU和内存资源请求和限制
HPA配置：水平Pod自动扩缩容
节点选择：节点选择器和节点亲和性
资源配额：命名空间资源配额管理
优先级：Pod优先级和抢占机制
示例配置：
---yaml
# flink-cluster.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: flink-jobmanager
spec:
replicas: 1
selector:
matchLabels:
app: flink
component: jobmanager
template:
metadata:
labels:
app: flink
component: jobmanager
spec:
containers:
- name: jobmanager
image: flink:1.16
ports:
- containerPort: 8081
env:
- name: JOB_MANAGER_RPC_ADDRESS
value: flink-jobmanager
resources:
requests:
memory: "1Gi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "1000m"
---
c.服务发现
Service服务：创建ClusterIP、NodePort、LoadBalancer服务
Ingress配置：使用Ingress暴露Web UI
DNS解析：Kubernetes DNS服务发现
健康检查：Pod健康检查和就绪检查
负载均衡：服务负载均衡和流量分发
b.K8s Application模式
a.作业作为K8s应用
自定义资源：Flink自定义资源定义（CRD）
应用生命周期：完整的Kubernetes应用生命周期
资源整合：深度集成Kubernetes资源管理
依赖管理：作业依赖的容器镜像管理
版本管理：应用版本管理和升级策略
b.自定义资源定义
FlinkDeployment：定义Flink部署资源
FlinkSessionJob：定义Session作业资源
FlinkStateBackend：状态后端配置
FlinkCheckpointing：检查点配置
CRD模式：Kubernetes Operator模式
示例配置：
---yaml
# flink-deployment.yaml
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: flink-job-example
spec:
image: flink:1.16
flinkVersion: v1_16
flinkConfiguration:
taskmanager.numberOfTaskSlots: "4"
state.backend: rocksdb
state.checkpoints.dir: "s3://checkpoints/"
state.savepoints.dir: "s3://savepoints/"
serviceAccount: flink
jobManager:
resource:
memory: "2048m"
cpu: 2
taskManager:
resource:
memory: "4096m"
cpu: 4
replicas: 3
job:
jarURI: local:///opt/flink/lib/job.jar
parallelism: 12
upgradeMode: savepoint
state: running
---
c.生命周期管理
应用启动：应用启动流程和状态检查
升级策略：滚动升级和蓝绿部署
回滚机制：应用回滚和版本管理
垃圾回收：无状态和有状态的垃圾回收
备份恢复：应用配置和状态的备份恢复
d.滚动更新
更新策略：滚动更新和分批更新策略
停机时间：最小化停机时间的更新策略
健康检查：更新过程中的健康检查
回滚机制：更新失败时的自动回滚
金丝雀发布：渐进式发布和流量控制
8.2 配置管理
01.核心配置
a.并行度配置
a.全局并行度
default.parallelism：设置作业的默认并行度
执行环境设置：通过ExecutionEnvironment配置
命令行参数：通过-p参数指定并行度
动态调整：运行时动态调整并行度
示例配置：
---yaml
# flink-conf.yaml
parallelism.default: 4
execution.checkpointing.interval: 5s
---
b.算子并行度
单个算子：通过setParallelism()方法设置
算子链：算子链中各算子的并行度配置
资源优化：根据算子特性优化并行度
分区策略：并行度与数据分区策略的关系
示例代码：
---
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置全局并行度
env.setParallelism(8);
// 设置单个算子并行度
DataStream<String> stream = env.addSource(sourceFunction)
.setParallelism(4)
.map(mapFunction)
.setParallelism(8)
.keyBy(keySelector)
.window(windowAssigner)
.aggregate(aggregateFunction)
.setParallelism(6);
---
c.槽位共享组
共享组概念：多个算子共享同一个TaskSlot
资源优化：通过Slot共享提高资源利用率
算子链配置：控制算子链的形成
资源隔离：不同共享组间的资源隔离
示例配置：
---
// 设置槽位共享组
DataStream<String> source = env.addSource(sourceFunction)
.setParallelism(4)
.slotSharingGroup("source-group");
DataStream<String> processed = source
.map(mapFunction)
.setParallelism(8)
.slotSharingGroup("processing-group");
// 禁用算子链
env.disableOperatorChaining();
// 控制算子链
source.map(mapFunction)
.startNewChain()
.filter(filterFunction);
---
b.内存配置
a.堆内存配置
JVM堆内存：JobManager和TaskManager的堆内存设置
GC调优：垃圾回收器选择和参数调优
内存分配：堆内存的分配策略和管理
OOM处理：内存溢出的预防和处理
示例配置：
---yaml
# 内存配置
jobmanager.memory.process.size: 1600m
jobmanager.memory.heap.size: 1024m
taskmanager.memory.process.size: 4096m
taskmanager.memory.heap.size: 2048m
taskmanager.memory.managed.fraction: 0.4
---
b.托管内存配置
托管内存：用于排序、哈希表、状态后端的内存
内存比例：托管内存占总内存的比例配置
内存分配：托管内存的分配和回收策略
性能优化：托管内存大小对性能的影响
示例配置：
---yaml
# 托管内存配置
taskmanager.memory.managed.size: 1024m
taskmanager.memory.managed.fraction: 0.4
# 或者使用具体大小
taskmanager.memory.managed.size: 512m
---
c.网络内存配置
网络缓冲区：网络数据传输的缓冲区配置
直接内存：网络直接内存的配置和管理
内存分配：网络内存的动态分配策略
反压处理：网络反压对内存的影响
示例配置：
---yaml
# 网络内存配置
taskmanager.memory.network.fraction: 0.1
taskmanager.memory.network.min: 64mb
taskmanager.memory.network.max: 1gb
---
d.内存调优策略
内存模型：Flink内存模型详细解析
监控指标：内存使用监控指标和分析
调优建议：不同场景的内存调优建议
故障排查：内存相关问题的排查方法
示例调优：
---yaml
# 生产环境内存调优示例
taskmanager.memory.process.size: 8192m
taskmanager.memory.heap.size: 4096m
taskmanager.memory.managed.size: 2048m
taskmanager.memory.network.fraction: 0.15
env.java.opts: "-XX:+UseG1GC -XX:MaxGCPauseMillis=200"
---
02.状态后端配置
a.状态存储配置
a.状态后端选择
MemoryStateBackend：基于内存的状态后端配置
FsStateBackend：基于文件系统的状态后端
RocksDBStateBackend：基于RocksDB的状态后端
状态后端对比：不同状态后端的特性对比和选择建议
示例配置：
---
// 代码中配置状态后端
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 内存状态后端
env.setStateBackend(new MemoryStateBackend());
// 文件系统状态后端
env.setStateBackend(new FsStateBackend("hdfs://namenode:port/flink/checkpoints"));
// RocksDB状态后端
env.setStateBackend(new RocksDBStateBackend("hdfs://namenode:port/flink/checkpoints", true));
---
b.检查点存储
检查点目录：检查点存储的目录配置
存储系统：支持HDFS、S3、本地文件系统等
检查点清理：检查点的自动清理策略
存储优化：检查点存储的性能优化
示例配置：
---yaml
# 检查点配置
state.backend: rocksdb
state.checkpoints.dir: hdfs://namenode:port/flink/checkpoints
state.checkpoints.num-retained: 20
state.savepoints.dir: hdfs://namenode:port/flink/savepoints
---
c.状态增量检查点
增量检查点：RocksDB增量检查点配置
本地增量：本地增量检查点机制
增量清理：增量检查点的清理策略
性能影响：增量检查点对性能的影响
示例配置：
---yaml
# RocksDB增量检查点
state.backend.incremental: true
state.backend.rocksdb.localdir: /local/flink/rocksdb
state.checkpoints.dir: s3://flink-bucket/checkpoints
---
d.本地恢复
本地恢复：启用本地恢复功能
恢复目录：本地恢复文件目录配置
恢复策略：本地恢复的策略和优先级
性能优化：本地恢复的性能优化
示例配置：
---yaml
# 本地恢复配置
state.local-recovery: true
state.backend.local-recovery: true
state.backend.rocksdb.localdir: /local/rocksdb/recovery
---
b.检查点配置
a.检查点间隔
时间间隔：检查点触发的时间间隔配置
最小间隔：检查点间的最小时间间隔
自适应间隔：基于负载的自适应检查点间隔
业务影响：检查点间隔对业务延迟的影响
示例配置：
---yaml
# 检查点间隔配置
execution.checkpointing.interval: 60s
execution.checkpointing.min-pause: 30s
execution.checkpointing.mode: EXACTLY_ONCE
---
b.超时配置
超时时间：单个检查点的超时时间配置
超时处理：检查点超时后的处理策略
重试机制：检查点失败的重试配置
监控告警：检查点超时的监控和告警
示例配置：
---yaml
# 检查点超时配置
execution.checkpointing.timeout: 10min
execution.checkpointing.tolerable-failure-number: 3
---
c.并发检查点
并发数量：同时进行的检查点数量限制
检查点队列：检查点队列的管理和优化
资源控制：并发检查点对资源的影响
性能优化：并发检查点的性能优化策略
示例配置：
---yaml
# 并发检查点配置
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.unaligned: false
---
d.保存点
保存点创建：手动创建保存点的方法
保存点格式：保存点的存储格式配置
保存点恢复：从保存点恢复作业的配置
版本升级：使用保存点进行版本升级
示例操作：
---bash
# 创建保存点
./bin/flink savepoint <jobId> <targetDirectory>
# 从保存点恢复作业
./bin/flink run -s <savepointPath> -p <parallelism> <jobJar>
# 取消作业时创建保存点
./bin/flink cancel -s <targetDirectory> <jobId>
---
03.网络配置
a.网络缓冲配置
a.网络缓冲区数量
缓冲区数量：每个TaskManager的网络缓冲区数量
内存大小：网络缓冲区占用的总内存大小
动态分配：网络缓冲区的动态分配机制
性能调优：缓冲区数量对网络性能的影响
示例配置：
---yaml
# 网络缓冲区配置
taskmanager.network.numberOfBuffers: 2048
taskmanager.network.memory.fraction: 0.1
taskmanager.network.memory.min: 64mb
taskmanager.network.memory.max: 1gb
---
b.内存分配策略
分配算法：网络内存的分配算法配置
缓冲区池：网络缓冲区池的管理策略
内存回收：网络内存的回收和复用机制
反压处理：网络反压对内存分配的影响
示例配置：
---yaml
# 网络内存分配策略
taskmanager.network.memory.buffers-per-channel: 2
taskmanager.network.memory.floating-buffers-per-gate: 16
taskmanager.network.memory.max: 1gb
---
b.容错配置
a.反压配置
反压监控：反压情况的监控和诊断
反压策略：反压的处理策略配置
网络配置：反压相关的网络参数调优
性能影响：反压对作业性能的影响
示例配置：
---yaml
# 反压相关配置
taskmanager.network.netty.num-arenas: 1
taskmanager.network.netty.transport: "nio"
---
b.重试策略
任务重启：失败任务的重启策略配置
重启次数：最大重启次数和重启间隔
重启延迟：重启的延迟时间和策略
故障恢复：各种故障场景的恢复策略
示例配置：
---yaml
# 重启策略配置
restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 10s
# 失败率重启策略
restart-strategy: failure-rate
restart-strategy.failure-rate.max-failures-per-interval: 5
restart-strategy.failure-rate.failure-rate-interval: 5min
restart-strategy.failure-rate.delay: 10s
---
c.超时配置
网络超时：网络通信的超时时间配置
心跳超时：心跳机制的超时时间设置
连接超时：连接建立的超时时间
请求超时：各种网络请求的超时配置
示例配置：
---yaml
# 超时配置
akka.ask.timeout: 60s
akka.tcp.timeout: 60s
heartbeat.timeout: 180s
heartbeat.interval: 30s
---
8.3 其他配置
01.监控指标
a.系统指标
a.CPU、内存使用率
CPU使用率：JVM进程和系统CPU使用率监控
内存使用率：堆内存、非堆内存的使用情况
垃圾回收：GC频率、停顿时间等GC指标
线程状态：线程数量、状态等线程相关指标
监控配置：
---yaml
# 指标报告配置
metrics.reporters: prom,jmx
metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
metrics.reporter.prom.port: 9249
metrics.reporter.jmx.class: org.apache.flink.metrics.jmx.JMXReporter
---
b.网络IO、磁盘IO
网络吞吐：网络读写速率和吞吐量监控
磁盘IO：磁盘读写速率和IOPS监控
文件句柄：文件句柄使用情况监控
网络连接：网络连接数和状态监控
指标示例：
---
// 自定义指标
public class MyOperator extends RichMapFunction<String, String> {
private transient Counter numRecords;
private transient Histogram recordSize;
@Override
public void open(Configuration parameters) {
numRecords = getRuntimeContext()
.getMetricGroup()
.counter("numRecords");
recordSize = getRuntimeContext()
.getMetricGroup()
.histogram("recordSize");
}
@Override
public String map(String value) throws Exception {
numRecords.inc();
recordSize.update(value.length());
return value.toUpperCase();
}
}
---
c.垃圾回收统计
GC监控：JVM垃圾回收的详细监控
内存泄漏：潜在的内存泄漏检测
堆内存分析：堆内存使用模式和趋势分析
GC调优：基于GC统计数据的调优建议
b.作业指标
a.吞吐量、延迟
数据吞吐：每秒处理的数据量统计
端到端延迟：从输入到输出的延迟测量
算子延迟：各个算子的处理延迟监控
SLA监控：延迟SLA的监控和告警
延迟监控配置：
---yaml
# 延迟监控配置
metrics.latency.granularity: operator
metrics.latency.history-size: 100
metrics.latency.interval: 30s
---
b.反压情况
反压检测：实时反压情况检测和报告
瓶颈识别：识别作业中的性能瓶颈
反压解决：反压问题的解决策略
资源调优：基于反压数据的资源调优
Web界面监控：
- 反压指示器：直观显示反压状态
- 背压监控：细粒度的背压监控
- 算子图：可视化的算子图和反压信息
c.检查点状态
检查点指标：检查点成功率、延迟等指标
状态大小：作业状态大小变化趋势
恢复时间：从检查点恢复的时间统计
检查点健康：检查点健康状态评估
监控指标：
---
// 检查点相关指标
- checkpoint.duration
- checkpoint.size
- checkpoint.alignment.duration
- checkpoint.num checkpoints
- checkpoint.num in progress
---
d.状态大小
状态监控：实时状态大小监控
增长趋势：状态大小的增长趋势分析
分布情况：状态在不同TaskManager间的分布
清理策略：基于状态监控的清理策略
02.Web界面
a.作业管理
a.作业提交和管理
作业提交：通过Web界面提交新作业
作业列表：显示所有作业的运行状态
作业详情：查看作业的详细配置和状态
作业控制：启动、停止、重启作业操作
主要功能：
- Overview页面：集群和作业概览
- Jobs页面：作业列表和状态
- Job详情：作业拓扑和执行计划
b.作业监控和调试
执行拓扑：可视化作业执行拓扑图
算子详情：查看各个算子的详细状态
实时指标：作业的实时性能指标显示
日志查看：在线查看作业执行日志
调试功能：
- Exception Tracking：异常跟踪和查看
- Checkpoint详情：检查点详细状态
- 反压监控：实时反压状态查看
- Timeline：作业执行时间线
c.保存点操作
保存点创建：通过Web界面创建保存点
保存点恢复：从保存点恢复作业
保存点管理：保存点的查看和删除
版本管理：保存点版本管理功能
操作界面：
- Savepoints页面：保存点管理界面
- 一键创建：快速创建保存点
- 恢复向导：保存点恢复向导
- 历史记录：保存点操作历史记录
b.集群管理
a.集群状态监控
集群概览：集群整体状态和资源使用情况
节点管理：TaskManager节点的状态和配置
资源分配：槽位分配和资源使用情况
性能指标：集群性能指标的实时监控
监控面板：
- Overview：集群概览仪表板
- TaskManagers：TaskManager管理页面
- JobManager：JobManager状态查看
- Configuration：集群配置查看
b.节点管理
节点状态：查看各个节点的运行状态
资源使用：节点资源使用情况监控
日志查看：节点日志的在线查看
故障处理：节点故障的诊断和处理
管理功能：
- 节点添加：动态添加TaskManager节点
- 节点移除：安全移除节点
- 资源调整：调整节点资源配置
- 重启操作：重启特定节点
c.配置查看
配置展示：集群配置参数的查看
配置对比：不同节点配置的对比
配置修改：配置参数的在线修改
配置验证：配置参数的验证和测试
配置管理：
- Global Configuration：全局配置查看
- Per-job Configuration：作业级配置
- System Metrics：系统级指标配置
- User Metrics：用户自定义指标
03.日志管理
a.日志配置
a.日志级别设置
级别配置：DEBUG、INFO、WARN、ERROR级别配置
包级别：针对特定包的日志级别设置
动态调整：运行时动态调整日志级别
日志过滤：基于条件过滤特定日志信息
配置示例：
---xml
<!-- log4j.properties -->
log4j.rootLogger=INFO, file, console
log4j.logger.org.apache.flink=INFO
log4j.logger.org.apache.flink.streaming.examples=WARN
log4j.logger.org.apache.flink.runtime.jobgraph=DEBUG
---
b.日志输出格式
格式模式：自定义日志输出格式模式
时间戳：日志时间戳格式配置
线程信息：线程名称和ID的日志格式
上下文信息：MDC上下文信息的日志输出
格式配置：
---xml
<!-- 日志格式配置 -->
log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n
log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} [%X{ jobId }] - %msg%n
---
c.日志轮转策略
大小轮转：基于文件大小的日志轮转
时间轮转：基于时间的日志轮转策略
轮转压缩：轮转日志文件的压缩配置
保留策略：日志文件的保留数量和期限
轮转配置：
---xml
<!-- 日志轮转配置 -->
log4j.appender.file=org.apache.log4j.RollingFileAppender
log4j.appender.file.File=logs/flink.log
log4j.appender.file.MaxFileSize=100MB
log4j.appender.file.MaxBackupIndex=10
log4j.appender.file.layout=org.apache.log4j.PatternLayout
---
b.日志分析
a.错误日志分析
错误收集：自动收集和分类错误日志
异常跟踪：异常堆栈的跟踪和分析
错误统计：错误发生频率和模式统计
告警机制：基于错误日志的告警机制
分析工具：
- 日志聚合：ELK Stack日志聚合
- 错误模式识别：基于机器学习的错误识别
- 根因分析：错误根因分析工具
- 趋势分析：错误趋势分析图表
b.性能日志分析
性能监控：性能相关的日志分析
瓶颈识别：基于日志的性能瓶颈识别
优化建议：性能优化建议的生成
趋势预测：性能趋势的预测和告警
分析方法：
- APM工具：集成APM性能监控
- 基准测试：性能基准测试和对比
- 容量规划：基于性能数据的容量规划
- 优化验证：优化效果的验证和评估
c.日志聚合和检索
日志收集：分布式日志的收集和传输
日志存储：大规模日志的存储方案
实时检索：实时日志的检索和查询
日志可视化：日志数据的可视化展示
技术方案：
---yaml
# ELK Stack配置
elasticsearch:
hosts: ["localhost:9200"]
index: "flink-logs-%{+yyyy.MM.dd}"
logstash:
input:
file:
path: "/path/to/flink/logs/*.log"
start_position: "beginning"
filter:
grok:
match: { "message": "%{TIMESTAMP_ISO8601:timestamp} [%DATA:thread] %LOGLEVEL:level %DATA:logger - %GREEDYDATA:message" }
output:
elasticsearch:
hosts: ["localhost:9200"]
kibana:
server.host: "localhost"
server.port: "5601"
---
8.4 性能优化
01.并行度优化
a.并行度设置策略
a.数据量和并行度关系
数据量评估：根据数据量确定合适的并行度范围
吞吐量要求：根据业务吞吐量需求调整并行度
延迟要求：低延迟场景下适当增加并行度
资源利用率：在满足性能要求的前提下提高资源利用率
计算公式：
---
// 并行度计算示例
int expectedTPS = 100000; // 期望吞吐量
int avgTPSPerSlot = 5000; // 每个槽位平均吞吐量
int optimalParallelism = (int) Math.ceil((double) expectedTPS / avgTPSPerSlot);
// 考虑资源限制
int availableSlots = totalTaskManagerSlots;
int finalParallelism = Math.min(optimalParallelism, availableSlots);
---
b.资源限制考虑
CPU资源：根据CPU核心数限制并行度
内存资源：考虑内存使用对并行度的影响
网络带宽：避免并行度过高导致网络瓶颈
存储IO：考虑存储系统IO能力限制
资源评估示例：
---
// 资源评估
int cpuCores = Runtime.getRuntime().availableProcessors();
int memoryPerTaskMB = 512; // 每个任务需要的内存
long totalMemoryMB = Runtime.getRuntime().maxMemory() / (1024 * 1024);
int memoryBasedParallelism = (int) (totalMemoryMB / memoryPerTaskMB);
int cpuBasedParallelism = cpuCores * 2; // 通常可以设置为CPU核心数的2-4倍
int recommendedParallelism = Math.min(memoryBasedParallelism, cpuBasedParallelism);
---
c.算子并行度调优
算子特性：根据算子特性调整并行度
数据倾斜：处理数据倾斜问题的并行度调整
背压监控：基于背压监控结果调整并行度
动态调整：支持运行时动态调整并行度
调优示例：
---
// 针对不同算子的并行度优化
DataStream<Event> source = env.addSource(kafkaSource)
.setParallelism(4) // 源算子并行度
.name("Kafka Source");
DataStream<Event> filtered = source.filter(filterFunction)
.setParallelism(8) // 过滤算子并行度
.name("Event Filter");
DataStream<AggregatedResult> aggregated = filtered
.keyBy(keySelector)
.window(windowAssigner)
.aggregate(aggregateFunction)
.setParallelism(12) // 聚合算子并行度
.name("Event Aggregation");
DataStream<Output> output = aggregated.map(outputMapper)
.setParallelism(4) // 输出算子并行度
.name("Output Mapper");
---
b.槽位共享优化
a.槽位共享组配置
共享组创建：创建合理的槽位共享组
算子链优化：优化算子链提高资源共享
资源隔离：不同共享组间的资源隔离
负载均衡：优化共享组内的负载均衡
配置示例：
---
// 槽位共享组配置
DataStream<String> source = env.addSource(sourceFunction)
.setParallelism(4)
.slotSharingGroup("source-group");
DataStream<String> processed = source
.map(mapFunction)
.filter(filterFunction)
.setParallelism(8)
.slotSharingGroup("processing-group");
// 相同共享组内的算子可以共享槽位
DataStream<String> result = processed
.keyBy(keySelector)
.window(windowAssigner)
.aggregate(aggregateFunction)
.setParallelism(6)
.slotSharingGroup("processing-group"); // 继续使用相同共享组
---
b.资源利用率优化
槽位数量：优化TaskManager的槽位数量配置
算子链长度：适当控制算子链长度
资源配置：优化JobManager和TaskManager资源配置
动态调整：支持动态调整槽位配置
配置参数：
---yaml
# flink-conf.yaml配置
taskmanager.numberOfTaskSlots: 8 # 每个TaskManager的槽位数
parallelism.default: 4 # 默认并行度
# 内存配置
taskmanager.memory.process.size: 8192m
taskmanager.memory.flink.size: 6400m
taskmanager.memory.network.fraction: 0.2
taskmanager.memory.managed.fraction: 0.4
---
c.链式操作优化
算子链控制：控制算子链的形成
链内优化：优化算子链内的操作
链间优化：优化算子链间的数据传输
性能监控：监控算子链的性能表现
链式操作示例：
---
// 算子链配置
DataStream<String> stream = env.addSource(sourceFunction)
.map(mapFunction) // 形成算子链
.filter(filterFunction) // 继续算子链
.flatMap(flatMapFunction); // 继续算子链
// 控制算子链
DataStream<String> chained = source
.map(mapFunction)
.startNewChain() // 开始新的算子链
.filter(filterFunction);
// 禁用算子链
env.disableOperatorChaining();
---
02.内存优化
a.内存管理配置
a.托管内存比例
比例设置：根据作业特点设置托管内存比例
状态内存：为有状态作业预留足够的托管内存
排序内存：为排序操作预留托管内存
动态调整：支持动态调整托管内存比例
配置示例：
---yaml
# 托管内存配置
taskmanager.memory.managed.fraction: 0.4 # 40%的托管内存
taskmanager.memory.managed.size: 2048m # 或者指定具体大小
# 状态相关配置
state.backend.rocksdb.memory.managed: true
state.backend.rocksdb.memory.fixed-per-slot: 256m
state.backend.rocksdb.memory.managed.consumer: 256m
---
b.网络缓冲区配置
缓冲区大小：优化网络缓冲区大小配置
缓冲区数量：调整网络缓冲区数量
动态分配：启用网络缓冲区动态分配
性能监控：监控网络缓冲区使用情况
配置参数：
---yaml
# 网络缓冲区配置
taskmanager.network.memory.fraction: 0.2
taskmanager.network.memory.min: 256mb
taskmanager.network.memory.max: 2gb
taskmanager.network.memory.buffers-per-channel: 4
taskmanager.network.memory.floating-buffers-per-gate: 16
---
c.批处理大小调整
Batch Size：调整批处理大小提高吞吐量
Batch Timeout：设置批处理超时时间
动态调整：根据网络状况动态调整批处理大小
性能测试：通过测试确定最优批处理大小
配置示例：
---
// Kafka连接器批处理配置
Properties props = new Properties();
props.setProperty("fetch.min.bytes", "1048576"); // 1MB
props.setProperty("fetch.max.bytes", "10485760"); // 10MB
props.setProperty("fetch.max.wait.ms", "500");
props.setProperty("max.poll.records", "1000");
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
"topic", new SimpleStringSchema(), props);
---
b.垃圾回收优化
a.GC策略选择
G1GC：推荐使用G1GC进行低延迟处理
GC参数：优化GC相关JVM参数
内存分配：优化JVM内存分配策略
监控GC：监控GC频率和停顿时间
JVM配置示例：
---yaml
# G1GC配置
env.java.opts: "-XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:G1HeapRegionSize=16m"
env.java.opts.jobmanager: "-XX:+UseG1GC -XX:MaxGCPauseMillis=100"
env.java.opts.taskmanager: "-XX:+UseG1GC -XX:MaxGCPauseMillis=200"
# 其他GC参数
env.java.opts: "-XX:+PrintGCDetails -XX:+PrintGCTimeStamps -Xloggc:gc.log"
env.java.opts: "-XX:+UseStringDeduplication -XX:+OptimizeStringConcat"
---
b.内存分配调优
堆内存：合理设置JVM堆内存大小
元空间：调整元空间大小避免溢出
直接内存：优化直接内存的使用
GC压力：减少GC压力的内存分配策略
内存调优示例：
---yaml
# 内存分配优化
jobmanager.memory.process.size: 2048m
jobmanager.memory.heap.size: 1536m
jobmanager.memory.off-heap.size: 256m
taskmanager.memory.process.size: 8192m
taskmanager.memory.heap.size: 4096m
taskmanager.memory.managed.size: 2048m
taskmanager.memory.network.fraction: 0.15
# JVM参数优化
env.java.opts: "-XX:+UseCompressedOops -XX:+UseCompressedClassPointers"
env.java.opts: "-XX:NewRatio=4 -XX:SurvivorRatio=8"
env.java.opts: "-XX:MaxDirectMemorySize=2048m"
---
c.对象复用策略
对象池：使用对象池减少对象创建
序列化：选择高效的序列化框架
预分配：预分配对象和内存空间
不可变对象：尽量使用不可变对象
对象复用示例：
---
// 对象池示例
public class EventObjectPool {
private final Queue<Event> pool = new ConcurrentLinkedQueue<>();
private final AtomicInteger created = new AtomicInteger(0);
public Event borrow() {
Event event = pool.poll();
if (event == null) {
event = new Event();
created.incrementAndGet();
}
return event;
}
public void returnObject(Event event) {
event.reset();
pool.offer(event);
}
}
// 在RichFunction中使用对象池
public class MyMapFunction extends RichMapFunction<Input, Output> {
private EventObjectPool objectPool;
@Override
public void open(Configuration parameters) {
objectPool = new EventObjectPool();
}
@Override
public Output map(Input value) throws Exception {
Event event = objectPool.borrow();
try {
// 使用event对象处理数据
return processWithEvent(event, value);
} finally {
objectPool.returnObject(event);
}
}
}
---
03.网络优化
a.网络配置优化
a.缓冲区大小调优
Buffer Size：调整网络缓冲区大小
Throughput：优化网络吞吐量配置
Latency：平衡吞吐量和延迟
Monitoring：监控网络性能指标
配置示例：
---yaml
# 网络缓冲区优化
taskmanager.network.memory.buffers-per-channel: 8
taskmanager.network.memory.floating-buffers-per-gate: 32
taskmanager.network.memory.fraction: 0.25
taskmanager.network.memory.min: 512mb
taskmanager.network.memory.max: 4gb
# Netty配置
taskmanager.network.netty.num-arenas: 2
taskmanager.network.netty.num-arenas: 2
taskmanager.network.netty.transport: "nio"
---
b.批处理策略
Batch Size：优化数据批处理大小
Batch Timeout：设置批处理超时
Adaptive Batch：自适应批处理策略
Performance Testing：通过测试确定最优参数
批处理配置：
---
// 自定义批处理策略
env.setBufferTimeout(Duration.ofMillis(100)); // 设置缓冲区超时
// Kafka消费者的批处理优化
Properties props = new Properties();
props.setProperty("fetch.min.bytes", "1MB");
props.setProperty("fetch.max.wait.ms", "500");
props.setProperty("max.poll.records", "5000");
// 自定义Source的批处理
public class BatchSource implements SourceFunction<String> {
private static final int BATCH_SIZE = 1000;
private final Queue<String> batch = new ArrayDeque<>(BATCH_SIZE);
@Override
public void run(SourceContext<String> ctx) throws Exception {
while (isRunning) {
// 批量收集数据
collectBatch();
if (batch.size() >= BATCH_SIZE) {
// 批量发送数据
batch.forEach(ctx::collect);
batch.clear();
}
Thread.sleep(10); // 控制批处理频率
}
}
}
---
c.压缩配置
Compression：启用网络传输压缩
Algorithm：选择合适的压缩算法
Trade-offs：平衡压缩率和CPU开销
Benchmarking：通过基准测试选择最优配置
压缩配置示例：
---yaml
# 网络压缩配置
taskmanager.network.compression: true
taskmanager.network.compression.algorithm: "lz4" # lz4, snappy, zlib
# Kafka压缩配置
producer.compression.type: "lz4"
producer.compression.level: 4
---
b.反压处理
a.反压检测机制
Backpressure Detection：启用反压检测
Sampling Rate：调整反压检测采样率
Metrics：监控反压相关指标
Alerting：设置反压告警阈值
反压监控：
---
// 反压指标监控
env.getConfig().setGlobalJobParameters(ParameterTool.fromArgs(new String[]{
"--metrics.backpressure.timeout", "10s",
"--metrics.backpressure.sample-interval", "5s"
}));
// Web界面中的反压监控
// - 绿色：无反压
// - 黄色：轻微反压
// - 红色：严重反压
---
b.反压缓解策略
Scale Out：通过增加并行度缓解反压
Resource Allocation：优化资源配置
Algorithm Optimization：优化算法复杂度
Data Skew Handling：处理数据倾斜问题
缓解策略示例：
---
// 数据倾斜处理
DataStream<Tuple2<String, Long>> skewedData = ...;
// 方法1：增加并行度
DataStream<Tuple2<String, Long>> scaled = skewedData
.keyBy(0)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sum(1)
.setParallelism(originalParallelism * 2);
// 方法2：局部聚合
DataStream<Tuple2<String, Long>> localAggregated = skewedData
.map(new RandomKeyMapper()) // 添加随机key
.keyBy(0)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sum(1)
.keyBy(0) // 移除随机key
.sum(1);
---
c.背压调优
Throughput Tuning：调整吞吐量配置
Buffer Size：优化缓冲区大小
Batch Processing：优化批处理策略
Load Balancing：改善负载均衡
调优最佳实践：
---yaml
# 反压调优配置
taskmanager.network.memory.fraction: 0.3
taskmanager.network.memory.max: 8gb
taskmanager.network.netty.num-arenas: 4
# 算子并行度调优
parallelism.default: 4
execution.buffer-timeout: 100ms
---
04.状态大小优化
a.状态数据类型选择
a.基本类型优先
Primitive Types：优先使用基本数据类型
Serialization：选择高效的序列化格式
Memory Layout：优化内存布局减少内存占用
Type Information：优化类型信息存储
类型选择示例：
---
// 使用基本类型
public class UserActivity {
private long userId; // long 优于Long
private int eventType; // int 优于Integer
private double amount; // double 优于Double
private long timestamp; // long 优于Long
private boolean isActive; // boolean 优于Boolean
}
// 自定义序列化器
public class UserActivitySerializer extends TypeSerializer<UserActivity> {
@Override
public void serialize(UserActivity record, DataOutputView target) throws IOException {
target.writeLong(record.getUserId());
target.writeInt(record.getEventType());
target.writeDouble(record.getAmount());
target.writeLong(record.getTimestamp());
target.writeBoolean(record.isIsActive());
}
}
---
b.数据结构优化
Efficient Collections：使用高效的数据结构
Sparse Data：对于稀疏数据使用特殊结构
Indexing：使用索引提高访问效率
Partitioning：优化数据分区策略
数据结构优化：
---
// 使用高效的数据结构
public class UserState {
// 使用原始类型集合
private final Long2ObjectOpenHashMap<UserProfile> userProfiles = new Long2ObjectOpenHashMap<>();
private final Long2LongOpenHashMap lastAccessTimes = new Long2LongOpenHashMap<>();
// 对于稀疏数据使用稀疏结构
private final SparseBitSet activeFeatures = new SparseBitSet();
// 使用Trove库的高性能集合
private final TObjectIntHashMap<String> featureCounts = new TObjectIntHashMap<>();
}
// 状态后端优化
stateDescriptor.enableTimeToLive(StateTtlConfig
.newBuilder(Time.hours(24))
.cleanupIncrementally(100, true)
.build());
---
c.序列化格式选择
Flink Type System：使用Flink类型系统
Custom Serialization：实现自定义序列化
Kryo vs Java：比较不同序列化格式
Serialization Performance：优化序列化性能
序列化优化：
---
// 注册类型信息提高序列化性能
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().registerTypeWithKryoSerializer(UserEvent.class, UserEventSerializer.class);
// 使用Kryo序列化
env.getConfig().enableObjectReuse();
env.getConfig().enableGenericTypes();
// 自定义类型信息
public class EventTypeInfo extends TypeInformation<UserEvent> {
@Override
public TypeSerializer<UserEvent> createSerializer(ExecutionConfig executionConfig) {
return new UserEventKryoSerializer();
}
}
---
b.状态清理策略
a.TTL配置
Time-to-Live：配置状态生存时间
Update Strategy：选择状态更新策略
Visibility Policy：配置TTL可见性策略
Cleanup Policy：配置清理策略
TTL配置示例：
---
// 状态TTL配置
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.hours(1)) // 1小时TTL
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.cleanupIncrementally(100, true) // 增量清理
.cleanupInRocksdbCompactFilter(1000) // RocksDB清理
.build();
ValueStateDescriptor<UserProfile> descriptor = new ValueStateDescriptor<>(
"userProfile",
TypeInformation.of(UserProfile.class)
);
descriptor.enableTimeToLive(ttlConfig);
---
b.状态压缩
Compression：启用状态压缩
Compression Algorithm：选择压缩算法
Compression Ratio：平衡压缩率和性能
Storage Optimization：优化存储使用
压缩配置：
---yaml
# RocksDB压缩配置
state.backend.rocksdb.compression.type: "lz4"
state.backend.rocksdb.compression.level: 4
state.backend.rocksdb.compaction.style: "universal"
state.backend.rocksdb.compaction.level.style: "compensated"
---
c.增量检查点
Incremental Checkpoints：启用增量检查点
Local State：优化本地状态管理
Checkpoint Size：减少检查点大小
Recovery Performance：提高恢复性能
增量检查点配置：
---yaml
# 增量检查点配置
state.backend: rocksdb
state.backend.incremental: true
state.backend.rocksdb.localdir: /opt/flink/rocksdb
state.checkpoints.dir: hdfs://namenode:port/flink/checkpoints
# 本地恢复配置
state.local-recovery: true
state.backend.local-recovery: true
---
05.检查点优化
a.检查点调优
a.检查点间隔优化
Interval Selection：选择合适的检查点间隔
Impact on Latency：考虑检查点对延迟的影响
Throughput Impact：考虑检查点对吞吐量的影响
Monitoring：监控检查点性能指标
间隔优化策略：
---yaml
# 检查点间隔优化
execution.checkpointing.interval: 60s # 基础间隔
execution.checkpointing.min-pause: 30s # 最小间隔
execution.checkpointing.timeout: 10min # 超时时间
# 根据业务特点调整
# 高延迟容忍：更长间隔，减少影响
# 低延迟要求：较短间隔，但可能影响性能
# 大状态：更长间隔，减少IO开销
# 小状态：较短间隔，快速故障恢复
---
b.并发检查点配置
Concurrent Checkpoints：配置并发检查点数量
Checkpoint Queue：管理检查点队列
Resource Management：优化检查点资源使用
Performance Trade-offs：平衡性能和资源使用
并发配置：
---yaml
# 并发检查点配置
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.unaligned: false
execution.checkpointing.tolerable-failure-number: 3
execution.checkpointing.alignment-timeout: 30s
---
c.异步检查点
Asynchronous Snapshots：启用异步检查点
Async IO：优化异步IO性能
Buffer Management：优化异步缓冲区管理
Performance Monitoring：监控异步检查点性能
异步配置：
---yaml
# 异步检查点配置
state.backend.async-snapshots: true
state.backend.rocksdb.checkpoint.thread.num: 4
state.backend.rocksdb.localdir: /opt/flink/rocksdb/local
---
b.保存点策略
a.保存点创建时机
Manual Savepoints：手动创建保存点的时机
Automatic Savepoints：自动保存点策略
Version Upgrades：版本升级时的保存点
Backup Strategy：备份策略和时机
创建策略：
---bash
# 手动创建保存点
./bin/flink savepoint <jobId> <targetDirectory>
# 在版本升级前创建保存点
./bin/flink stop --savepointPath <savepointDir> <jobId>
# 定期自动创建保存点（通过作业逻辑）
public class PeriodicSavepointTrigger extends RichFunction {
@Override
public void open(Configuration parameters) {
// 设置定时器定期创建保存点
getRuntimeContext().getTimerService()
.registerProcessingTimeTimer(
System.currentTimeMillis() + TimeUnit.HOURS.toMillis(6)
);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) {
// 触发保存点
triggerSavepoint();
// 注册下一个定时器
getRuntimeContext().getTimerService()
.registerProcessingTimeTimer(
timestamp + TimeUnit.HOURS.toMillis(6)
);
}
}
---
b.保存点清理策略
Retention Policy：配置保存点保留策略
Automatic Cleanup：自动清理过期保存点
Storage Management：优化存储使用
Backup Retention：备份保留策略
清理配置：
---yaml
# 保存点清理配置
state.savepoints.dir: hdfs://namenode:port/flink/savepoints
state.checkpoints.num-retained: 20 # 保留20个检查点
state.savepoints.num-retained: 5 # 保留5个保存点
# 定期清理脚本
# */6 * * * * find /flink/savepoints -type d -mtime +7 -exec rm -rf {} \;
---
c.增量保存点
Incremental Savepoints：启用增量保存点
Local State：优化本地状态管理
Storage Efficiency：提高存储效率
Recovery Performance：提高恢复性能
增量配置：
---yaml
# 增量保存点配置
state.backend.incremental: true
state.backend.rocksdb.localdir: /opt/flink/rocksdb
state.savepoints.dir: s3://flink-bucket/savepoints
state.checkpoints.dir: s3://flink-bucket/checkpoints
---
06.状态后端选择
a.不同场景选择
a.内存状态后端适用场景
Development：开发和测试环境
Small State：小状态作业
Low Throughput：低吞吐量应用
Quick Setup：快速原型和验证
Memory State Backend配置：
---
// 内存状态后端配置
env.setStateBackend(new MemoryStateBackend());
env.enableCheckpointing(5000); // 5秒检查点间隔
// 适用于小状态作业
env.setStateBackend(new MemoryStateBackend(
1024 * 1024 * 1024, // 1GB最大状态大小
false // 异步快照
));
---
b.文件系统状态后端适用场景
Medium State：中等大小状态
Production：生产环境
Reliability：需要可靠性保证
File System：基于文件系统的存储
FsState Backend配置：
---
// 文件系统状态后端配置
env.setStateBackend(new FsStateBackend(
"hdfs://namenode:port/flink/checkpoints", // 检查点目录
true // 异步快照
));
// S3文件系统配置
env.setStateBackend(new FsStateBackend(
"s3a://flink-bucket/checkpoints",
true
));
---
c.RocksDB状态后端适用场景
Large State：大状态作业
High Throughput：高吞吐量应用
Production Ready：生产级应用
Incremental Checkpoints：增量检查点需求
RocksDB State Backend配置：
---
// RocksDB状态后端配置
env.setStateBackend(new RocksDBStateBackend(
"hdfs://namenode:port/flink/checkpoints", // 检查点目录
true, // 增量检查点
true // 异步快照
));
// RocksDB本地目录配置
config.setString("state.backend.rocksdb.localdir", "/opt/flink/rocksdb");
// RocksDB优化参数
config.setString("state.backend.rocksdb.memory.managed", "true");
config.setString("state.backend.rocksdb.memory.fixed-per-slot", "256m");
---
b.状态后端优化
a.本地恢复配置
Local Recovery：启用本地恢复
Recovery Directory：配置恢复目录
Recovery Strategy：优化恢复策略
Performance Improvement：提高恢复性能
本地恢复配置：
---yaml
# 本地恢复配置
state.local-recovery: true
state.backend.local-recovery: true
state.backend.rocksdb.localdir: /opt/flink/rocksdb/recovery
# 本地恢复目录配置
state.backend.rocksdb.localdir: /opt/flink/rocksdb/local1,/opt/flink/rocksdb/local2
---
b.增量检查点配置
Incremental Snapshots：启用增量快照
Local RocksDB：优化本地RocksDB配置
Upload Strategy：优化上传策略
Storage Optimization：优化存储使用
增量检查点优化：
---yaml
# 增量检查点优化配置
state.backend.incremental: true
state.backend.rocksdb.localdir: /opt/flink/rocksdb
state.backend.rocksdb.checkpoint.thread.num: 8
state.backend.rocksdb.write-buffer.size: 64mb
state.backend.rocksdb.compaction.style: universal
---
c.状态访问优化
Access Pattern：优化状态访问模式
Cache Strategy：使用缓存策略
Serialization：优化状态序列化
Memory Management：优化内存管理
访问优化示例：
---
// 状态访问优化
public class UserBehaviorProcessor extends KeyedProcessFunction<Long, UserEvent, Output> {
private transient ValueState<UserProfile> userProfileState;
private transient MapState<String, Integer> featureCountState;
@Override
public void open(Configuration parameters) {
// 状态描述符配置
ValueStateDescriptor<UserProfile> profileDesc = new ValueStateDescriptor<>(
"userProfile",
TypeInformation.of(UserProfile.class)
);
// 启用TTL减少状态大小
profileDesc.enableTimeToLive(StateTtlConfig
.newBuilder(Time.days(30))
.cleanupIncrementally(100, true)
.build());
userProfileState = getRuntimeContext().getState(profileDesc);
// MapState优化配置
MapStateDescriptor<String, Integer> featureDesc = new MapStateDescriptor<>(
"featureCount",
TypeInformation.of(String.class),
TypeInformation.of(Integer.class)
);
featureCountState = getRuntimeContext().getMapState(featureDesc);
}
@Override
public void processElement(UserEvent event, Context ctx, Collector<Output> out) throws Exception {
// 批量访问状态，减少状态访问次数
UserProfile profile = userProfileState.value();
if (profile == null) {
profile = new ProfileBuilder().build();
}
// 批量更新状态
profile.updateFromEvent(event);
userProfileState.update(profile);
// 使用迭代器批量访问MapState
for (Map.Entry<String, Integer> entry : featureCountState.entries()) {
// 处理特征计数
}
}
}
---
9 实战案例
9.1 实时数据分析
01.用户行为分析
a.需求分析
实时用户画像：构建用户特征标签，支持个性化推荐
行为模式识别：识别用户访问模式、购买意图等
实时推荐：基于用户行为实时调整推荐策略
b.技术架构
数据源 → Kafka → Flink → Redis/MySQL → 前端展示
c.代码实现
a.数据模型定义
// 用户行为事件
public class UserEvent {
private String userId;
private String eventType; // "click", "view", "purchase"
private String itemId;
private long timestamp;
private Map<String, Object> properties;
// 构造函数、getter、setter省略
}
// 用户画像
public class UserProfile {
private String userId;
private Map<String, Double> preferences; // 兴趣偏好
private Map<String, Integer> behaviors; // 行为统计
private long lastUpdateTime;
// 构造函数、getter、setter省略
}
b.Flink处理逻辑
// 1. 从Kafka消费用户行为数据
DataStream<UserEvent> eventStream = env.addSource(
new FlinkKafkaConsumer<>(
"user-events",
new UserEventDeserializer(),
kafkaProps
)
);
// 2. 用户行为统计
KeyedStream<UserEvent, String> keyedByUser = eventStream
.keyBy(UserEvent::getUserId);
// 3. 实时计算用户画像
DataStream<UserProfile> userProfiles = keyedByUser
.process(new UserProfileProcessFunction());
// 4. 实时推荐
DataStream<Recommendation> recommendations = userProfiles
.connect(eventStream.keyBy(UserEvent::getUserId))
.flatMap(new RecommendationFunction());
// 5. 输出结果
recommendations.addSink(
new RedisSink<>(redisConfig, new RedisRecommendationMapper())
);
c.用户画像处理函数
public class UserProfileProcessFunction
extends KeyedProcessFunction<String, UserEvent, UserProfile> {
private ValueState<UserProfile> profileState;
@Override
public void open(Configuration parameters) {
profileState = getRuntimeContext().getState(
new ValueStateDescriptor<>("profile", UserProfile.class)
);
}
@Override
public void processElement(UserEvent event, Context ctx,
Collector<UserProfile> out) throws Exception {
UserProfile profile = profileState.value();
if (profile == null) {
profile = new UserProfile();
profile.setUserId(event.getUserId());
}
// 更新行为统计
profile.updateBehavior(event.getEventType());
// 更新兴趣偏好
profile.updatePreference(event.getItemId(), event.getEventType());
profile.setLastUpdateTime(ctx.timerService().currentProcessingTime());
profileState.update(profile);
out.collect(profile);
}
}
d.配置文件
# application.yml
flink:
kafka:
bootstrap-servers: localhost:9092
group-id: user-behavior-analysis
topic: user-events
redis:
host: localhost
port: 6379
database: 0
checkpoint:
interval: 60000 # 1分钟
timeout: 300000 # 5分钟
02.实时监控告警
a.监控指标
// 系统性能指标
public class SystemMetrics {
private double cpuUsage;
private double memoryUsage;
private long diskIO;
private long networkIO;
private int activeConnections;
}
// 业务指标
public class BusinessMetrics {
private long orderCount;
private double orderAmount;
private int activeUsers;
private double responseTime;
private int errorCount;
}
b.告警处理逻辑
// 1. 指标数据源
DataStream<SystemMetrics> systemMetrics = env
.addSource(new SystemMetricsSource());
DataStream<BusinessMetrics> businessMetrics = env
.addSource(new BusinessMetricsSource());
// 2. 阈值告警
Pattern<SystemMetrics, ?> thresholdPattern = Pattern.<SystemMetrics>begin("threshold")
.where(metrics -> metrics.getCpuUsage() > 0.8 || metrics.getMemoryUsage() > 0.9);
DataStream<Alert> thresholdAlerts = CEP.pattern(systemMetrics, thresholdPattern)
.select(pattern -> {
SystemMetrics metrics = pattern.get("threshold");
return new Alert(
AlertType.THRESHOLD,
"系统资源使用率过高",
metrics
);
});
// 3. 趋势告警
KeyedStream<BusinessMetrics, String> keyedMetrics = businessMetrics
.keyBy(metrics -> "global");
DataStream<Alert> trendAlerts = keyedMetrics
.process(new TrendDetectionFunction());
// 4. 告警聚合和去重
DataStream<Alert> allAlerts = thresholdAlerts
.union(trendAlerts)
.keyBy(Alert::getType)
.process(new AlertAggregationFunction());
// 5. 告警输出
allAlerts.addSink(new AlertNotificationSink());
c.趋势检测函数
public class TrendDetectionFunction
extends KeyedProcessFunction<String, BusinessMetrics, Alert> {
private ListState<BusinessMetrics> metricsHistory;
private static final int HISTORY_SIZE = 10;
@Override
public void open(Configuration parameters) {
metricsHistory = getRuntimeContext().getListState(
new ListStateDescriptor<>("metrics", BusinessMetrics.class)
);
}
@Override
public void processElement(BusinessMetrics metrics, Context ctx,
Collector<Alert> out) throws Exception {
List<BusinessMetrics> history = new ArrayList<>();
for (BusinessMetrics m : metricsHistory.get()) {
history.add(m);
}
if (history.size() >= HISTORY_SIZE) {
// 检测错误率上升趋势
if (isErrorRateIncreasing(history)) {
out.collect(new Alert(
AlertType.TREND,
"错误率持续上升",
metrics
));
}
// 检测响应时间上升趋势
if (isResponseTimeIncreasing(history)) {
out.collect(new Alert(
AlertType.TREND,
"响应时间持续增加",
metrics
));
}
history.remove(0);
}
history.add(metrics);
metricsHistory.update(history);
}
}
d.告警通知实现
public class AlertNotificationSink implements SinkFunction<Alert> {
@Override
public void invoke(Alert alert, Context context) throws Exception {
// 邮件通知
sendEmailNotification(alert);
// 钉钉/企业微信通知
sendIMNotification(alert);
// 短信通知（紧急告警）
if (alert.getSeverity() == AlertSeverity.CRITICAL) {
sendSMSNotification(alert);
}
// 写入告警日志
logAlert(alert);
}
}
9.2 事件驱动应用
01.实时风控系统
a.风控需求
实时风险评估：毫秒级响应，识别潜在风险
规则引擎：支持动态配置风控规则
机器学习模型：使用历史数据进行智能风控决策
b.系统架构
交易事件 → Kafka → Flink → 规则引擎 → ML模型 → 风险决策 → 拦截/通过
↓
特征存储(Redis)
c.核心代码实现
a.风控事件模型
// 风险评估事件
public class RiskEvent {
private String eventId;
private String userId;
private String deviceId;
private String transactionType; // "payment", "transfer", "login"
private double amount;
private String location;
private long timestamp;
private Map<String, Object> features;
}
// 风险决策结果
public class RiskDecision {
private String eventId;
private RiskLevel riskLevel;
private String riskReason;
private Map<String, Object> featureValues;
private boolean blockTransaction;
}
b.实时风控处理逻辑
// 1. 风险事件流
DataStream<RiskEvent> riskEvents = env
.addSource(new FlinkKafkaConsumer<>(
"risk-events",
new RiskEventDeserializer(),
kafkaProps
));
// 2. 特征计算
DataStream<RiskFeature> features = riskEvents
.keyBy(RiskEvent::getUserId)
.process(new RiskFeatureCalculationFunction());
// 3. 规则引擎处理
DataStream<RiskResult> ruleResults = features
.process(new RuleEngineProcessFunction());
// 4. 机器学习模型评分
DataStream<RiskScore> mlScores = features
.process(new MLScoringFunction());
// 5. 综合风险决策
DataStream<RiskDecision> decisions = ruleResults
.connect(mlScores)
.keyBy(RiskResult::getEventId, RiskScore::getEventId)
.flatMap(new DecisionMakingFunction());
// 6. 输出决策结果
decisions.addSink(new RiskDecisionSink());
c.风险特征计算
public class RiskFeatureCalculationFunction
extends KeyedProcessFunction<String, RiskEvent, RiskFeature> {
private ValueState<UserBehavior> userBehaviorState;
private MapState<String, Long> transactionCountState;
@Override
public void open(Configuration parameters) {
// 用户行为状态
userBehaviorState = getRuntimeContext().getState(
new ValueStateDescriptor<>("userBehavior", UserBehavior.class)
);
// 交易计数状态
transactionCountState = getRuntimeContext().getMapState(
new MapStateDescriptor<>("transactionCount", String.class, Long.class)
);
}
@Override
public void processElement(RiskEvent event, Context ctx,
Collector<RiskFeature> out) throws Exception {
UserBehavior behavior = userBehaviorState.value();
if (behavior == null) {
behavior = new UserBehavior(event.getUserId());
}
// 计算各种风险特征
RiskFeature feature = new RiskFeature(event.getEventId());
// 1. 交易频率特征
long recentTxnCount = getRecentTransactionCount(ctx.timestamp());
feature.setTransactionFrequency(recentTxnCount);
// 2. 金额异常特征
feature.setAmountAnomaly(calculateAmountAnomaly(event, behavior));
// 3. 地理位置异常特征
feature.setLocationAnomaly(calculateLocationAnomaly(event, behavior));
// 4. 设备风险特征
feature.setDeviceRisk(calculateDeviceRisk(event));
// 5. 时间模式特征
feature.setTimePatternAnomaly(calculateTimePattern(event, behavior));
// 更新用户行为状态
behavior.updateBehavior(event);
userBehaviorState.update(behavior);
out.collect(feature);
}
}
d.规则引擎实现
public class RuleEngineProcessFunction
extends KeyedProcessFunction<String, RiskFeature, RiskResult> {
private List<RiskRule> rules;
@Override
public void open(Configuration parameters) {
// 加载风控规则（可动态更新）
loadRiskRules();
}
@Override
public void processElement(RiskFeature feature, Context ctx,
Collector<RiskResult> out) throws Exception {
RiskResult result = new RiskResult(feature.getEventId());
int riskScore = 0;
for (RiskRule rule : rules) {
if (rule.matches(feature)) {
riskScore += rule.getScore();
result.addRuleHit(rule.getName(), rule.getScore());
}
}
result.setRiskScore(riskScore);
result.setRiskLevel(determineRiskLevel(riskScore));
out.collect(result);
}
private RiskLevel determineRiskLevel(int score) {
if (score >= 80) return RiskLevel.HIGH;
if (score >= 50) return RiskLevel.MEDIUM;
return RiskLevel.LOW;
}
}
e.动态规则配置
# risk-rules.yml
rules:
- name: "high_amount_rule"
description: "大额交易风险"
conditions:
- field: "amount"
operator: ">"
value: 10000
score: 30
enabled: true
- name: "frequency_rule"
description: "高频交易风险"
conditions:
- field: "transactionFrequency"
operator: ">"
value: 10
score: 25
enabled: true
- name: "location_anomaly_rule"
description: "地理位置异常"
conditions:
- field: "locationAnomaly"
operator: "=="
value: true
score: 20
enabled: true
02.物联网数据处理
a.数据特点
大规模设备连接：百万级设备同时在线
高频数据产生：每秒产生数百万条数据点
实时性要求高：延迟需控制在秒级以内
b.技术架构
传感器设备 → MQTT Broker → Kafka → Flink → 时序数据库 → 监控平台
↓
异常检测 → 告警系统
c.IoT数据模型
// 设备传感器数据
public class SensorData {
private String deviceId;
private String sensorType; // "temperature", "humidity", "pressure"
private double value;
private String unit;
private long timestamp;
private Map<String, String> metadata;
}
// 设备状态信息
public class DeviceStatus {
private String deviceId;
private DeviceState state; // "online", "offline", "error"
private double batteryLevel;
private double signalStrength;
private long lastSeenTime;
}
d.实时处理逻辑
// 1. 多源数据接入
DataStream<SensorData> sensorStream = env
.addSource(new FlinkKafkaConsumer<>(
"iot-sensor-data",
new SensorDataDeserializer(),
kafkaProps
))
.name("sensor-data-source");
DataStream<DeviceStatus> deviceStatusStream = env
.addSource(new FlinkKafkaConsumer<>(
"iot-device-status",
new DeviceStatusDeserializer(),
kafkaProps
))
.name("device-status-source");
// 2. 数据清洗和预处理
DataStream<SensorData> cleanedSensorData = sensorStream
.filter(data -> data.getValue() != null && !Double.isNaN(data.getValue()))
.keyBy(SensorData::getDeviceId)
.process(new DataValidationFunction())
.name("data-validation");
// 3. 实时异常检测
DataStream<AnomalyAlert> anomalyAlerts = cleanedSensorData
.keyBy(data -> data.getDeviceId() + "_" + data.getSensorType())
.process(new AnomalyDetectionFunction())
.name("anomaly-detection");
// 4. 设备状态监控
DataStream<DeviceAlert> deviceAlerts = deviceStatusStream
.keyBy(DeviceStatus::getDeviceId)
.process(new DeviceHealthMonitorFunction())
.name("device-health-monitor");
// 5. 聚合统计
DataStream<DeviceStats> deviceStats = cleanedSensorData
.keyBy(SensorData::getDeviceId)
.window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
.aggregate(new DeviceStatsAggregator())
.name("device-stats-aggregation");
// 6. 结果输出
anomalyAlerts.addSink(new AlertSink());
deviceAlerts.addSink(new AlertSink());
deviceStats.addSink(new InfluxDBSink());
e.异常检测算法实现
public class AnomalyDetectionFunction
extends KeyedProcessFunction<String, SensorData, AnomalyAlert> {
private ValueState<MovingAverageCalculator> movingAverageState;
private ValueState<StandardDeviationCalculator> stdDevState;
private ListState<SensorData> recentDataState;
private static final int WINDOW_SIZE = 100;
private static final double ANOMALY_THRESHOLD = 3.0; // 3σ原则
@Override
public void open(Configuration parameters) {
movingAverageState = getRuntimeContext().getState(
new ValueStateDescriptor<>("movingAvg", MovingAverageCalculator.class)
);
stdDevState = getRuntimeContext().getState(
new ValueStateDescriptor<>("stdDev", StandardDeviationCalculator.class)
);
recentDataState = getRuntimeContext().getListState(
new ListStateDescriptor<>("recentData", SensorData.class)
);
}
@Override
public void processElement(SensorData data, Context ctx,
Collector<AnomalyAlert> out) throws Exception {
// 维护滑动窗口数据
List<SensorData> recentData = getRecentData(data);
recentDataState.update(recentData);
if (recentData.size() >= 20) { // 至少需要20个数据点
// 计算统计指标
double mean = calculateMean(recentData);
double stdDev = calculateStdDev(recentData, mean);
// 检测异常（使用Z-score方法）
double zScore = Math.abs((data.getValue() - mean) / stdDev);
if (zScore > ANOMALY_THRESHOLD) {
AnomalyAlert alert = new AnomalyAlert();
alert.setDeviceId(data.getDeviceId());
alert.setSensorType(data.getSensorType());
alert.setAnomalyValue(data.getValue());
alert.setExpectedValue(mean);
alert.setZScore(zScore);
alert.setTimestamp(data.getTimestamp());
alert.setSeverity(determineSeverity(zScore));
out.collect(alert);
}
}
}
private List<SensorData> getRecentData(SensorData newData) throws Exception {
List<SensorData> result = new ArrayList<>();
// 获取历史数据
for (SensorData data : recentDataState.get()) {
result.add(data);
}
// 添加新数据
result.add(newData);
// 保持窗口大小
while (result.size() > WINDOW_SIZE) {
result.remove(0);
}
return result;
}
private AnomalySeverity determineSeverity(double zScore) {
if (zScore > 5.0) return AnomalySeverity.CRITICAL;
if (zScore > 4.0) return AnomalySeverity.HIGH;
if (zScore > 3.0) return AnomalySeverity.MEDIUM;
return AnomalySeverity.LOW;
}
}
f.设备健康监控
public class DeviceHealthMonitorFunction
extends KeyedProcessFunction<String, DeviceStatus, DeviceAlert> {
private ValueState<DeviceStatus> lastStatusState;
private ValueState<Long> lastSeenTimeState;
@Override
public void open(Configuration parameters) {
lastStatusState = getRuntimeContext().getState(
new ValueStateDescriptor<>("lastStatus", DeviceStatus.class)
);
lastSeenTimeState = getRuntimeContext().getState(
new ValueStateDescriptor<>("lastSeen", Long.class)
);
}
@Override
public void processElement(DeviceStatus status, Context ctx,
Collector<DeviceAlert> out) throws Exception {
DeviceStatus lastStatus = lastStatusState.value();
// 检查设备离线
if (lastStatus != null &&
lastStatus.getState() == DeviceState.ONLINE &&
status.getState() == DeviceState.OFFLINE) {
DeviceAlert alert = new DeviceAlert();
alert.setDeviceId(status.getDeviceId());
alert.setAlertType(AlertType.DEVICE_OFFLINE);
alert.setSeverity(AlertSeverity.HIGH);
alert.setMessage("设备突然离线");
alert.setTimestamp(System.currentTimeMillis());
out.collect(alert);
}
// 检查电池电量低
if (status.getBatteryLevel() < 10.0) {
DeviceAlert alert = new DeviceAlert();
alert.setDeviceId(status.getDeviceId());
alert.setAlertType(AlertType.LOW_BATTERY);
alert.setSeverity(AlertSeverity.MEDIUM);
alert.setMessage("设备电量不足: " + status.getBatteryLevel() + "%");
alert.setTimestamp(System.currentTimeMillis());
out.collect(alert);
}
// 检查信号强度弱
if (status.getSignalStrength() < -80) {
DeviceAlert alert = new DeviceAlert();
alert.setDeviceId(status.getDeviceId());
alert.setAlertType(AlertType.WEAK_SIGNAL);
alert.setSeverity(AlertSeverity.LOW);
alert.setMessage("设备信号强度弱: " + status.getSignalStrength() + "dBm");
alert.setTimestamp(System.currentTimeMillis());
out.collect(alert);
}
// 设置离线检测定时器
ctx.timerService().registerProcessingTimeTimer(
System.currentTimeMillis() + 300000 // 5分钟后检查
);
lastStatusState.update(status);
lastSeenTimeState.update(System.currentTimeMillis());
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx,
Collector<DeviceAlert> out) throws Exception {
Long lastSeen = lastSeenTimeState.value();
if (lastSeen != null && (System.currentTimeMillis() - lastSeen) > 300000) {
DeviceAlert alert = new DeviceAlert();
alert.setDeviceId(ctx.getCurrentKey());
alert.setAlertType(AlertType.DEVICE_MISSING);
alert.setSeverity(AlertSeverity.CRITICAL);
alert.setMessage("设备长时间未上报数据");
alert.setTimestamp(System.currentTimeMillis());
out.collect(alert);
}
}
}
9.3 ETL处理
01.实时数据清洗
a.数据质量问题
数据格式不一致：日期格式、数值精度、编码格式等差异
数据缺失：关键字段为空或缺失
数据异常：超出正常范围的数值、格式错误等
b.清洗策略架构
---
原始数据 → 数据校验 → 格式转换 → 异常检测 → 数据补全 → 清洗数据
↓ ↓ ↓ ↓ ↓
错误日志 转换日志 异常日志 补全日志 质量报告
---
c.代码实现
a.数据质量规则定义
---
// 数据质量规则
public class DataQualityRule {
private String ruleName;
private String fieldName;
private RuleType ruleType; // NOT_NULL, RANGE, FORMAT, PATTERN
private Object ruleValue;
private String errorMessage;
private QualityLevel qualityLevel;
}
// 数据质量检查结果
public class QualityCheckResult {
private String recordId;
private boolean isValid;
private List<String> violations;
private QualityScore qualityScore;
private Map<String, Object> originalData;
private Map<String, Object> cleanedData;
}
---
b.实时数据清洗处理
---
// 1. 原始数据流
DataStream<RawData> rawDataStream = env
.addSource(new FlinkKafkaConsumer<>(
"raw-data-topic",
new RawDataDeserializer(),
kafkaProps
))
.name("raw-data-source");
// 2. 数据质量检查
DataStream<QualityCheckResult> qualityResults = rawDataStream
.keyBy(RawData::getRecordId)
.process(new DataQualityCheckFunction())
.name("data-quality-check");
// 3. 数据清洗处理
DataStream<CleanData> cleanDataStream = qualityResults
.filter(QualityCheckResult::isValid)
.process(new DataCleaningFunction())
.name("data-cleaning");
// 4. 异常数据处理
DataStream<ExceptionData> exceptionStream = qualityResults
.filter(result -> !result.isValid())
.process(new ExceptionHandlingFunction())
.name("exception-handling");
// 5. 数据质量监控
DataStream<QualityMetrics> qualityMetrics = qualityResults
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.aggregate(new QualityMetricsAggregator())
.name("quality-metrics");
// 6. 结果输出
cleanDataStream.addSink(new FlinkKafkaProducer<>(
"clean-data-topic",
new CleanDataSerializer(),
kafkaProps
));
exceptionStream.addSink(new ExceptionDataSink());
qualityMetrics.addSink(new QualityMetricsSink());
---
c.数据质量检查函数
---
public class DataQualityCheckFunction
extends KeyedProcessFunction<String, RawData, QualityCheckResult> {
private ListState<DataQualityRule> rulesState;
@Override
public void open(Configuration parameters) {
rulesState = getRuntimeContext().getListState(
new ListStateDescriptor<>("rules", DataQualityRule.class)
);
// 加载数据质量规则
loadDataQualityRules();
}
@Override
public void processElement(RawData data, Context ctx,
Collector<QualityCheckResult> out) throws Exception {
QualityCheckResult result = new QualityCheckResult();
result.setRecordId(data.getRecordId());
result.setOriginalData(data.getData());
List<String> violations = new ArrayList<>();
Map<String, Object> cleanedData = new HashMap<>(data.getData());
for (DataQualityRule rule : getRules()) {
if (!checkRule(data, rule)) {
violations.add(rule.getErrorMessage());
// 尝试自动修复
if (rule.getQualityLevel() == QualityLevel.MEDIUM) {
tryAutoFix(cleanedData, rule);
}
}
}
result.setViolations(violations);
result.setValid(violations.isEmpty());
result.setCleanedData(cleanedData);
result.setQualityScore(calculateQualityScore(violations));
out.collect(result);
}
private boolean checkRule(RawData data, DataQualityRule rule) {
Object value = data.getData().get(rule.getFieldName());
switch (rule.getRuleType()) {
case NOT_NULL:
return value != null && !value.toString().trim().isEmpty();
case RANGE:
if (value instanceof Number) {
double numValue = ((Number) value).doubleValue();
Range<Double> range = (Range<Double>) rule.getRuleValue();
return range.contains(numValue);
}
return false;
case FORMAT:
return value.toString().matches((String) rule.getRuleValue());
case PATTERN:
Pattern pattern = Pattern.compile((String) rule.getRuleValue());
return pattern.matcher(value.toString()).matches();
default:
return true;
}
}
private void tryAutoFix(Map<String, Object> data, DataQualityRule rule) {
switch (rule.getRuleType()) {
case FORMAT:
// 日期格式标准化
if (rule.getFieldName().equals("date")) {
String dateStr = (String) data.get(rule.getFieldName());
data.put(rule.getFieldName(), standardizeDate(dateStr));
}
break;
case NOT_NULL:
// 空值补全
if (data.get(rule.getFieldName()) == null) {
data.put(rule.getFieldName(), getDefaultValue(rule.getFieldName()));
}
break;
}
}
}
---
d.配置文件
---
# data-quality-rules.yml
quality_rules:
- rule_name: "user_id_not_null"
field_name: "user_id"
rule_type: "NOT_NULL"
quality_level: "HIGH"
error_message: "用户ID不能为空"
- rule_name: "amount_range"
field_name: "amount"
rule_type: "RANGE"
rule_value: {"min": 0.0, "max": 1000000.0}
quality_level: "MEDIUM"
error_message: "交易金额必须在0-1000000之间"
- rule_name: "email_format"
field_name: "email"
rule_type: "FORMAT"
rule_value: "^[\\w-\\.]+@([\\w-]+\\.)+[\\w-]{2,4}$"
quality_level: "LOW"
error_message: "邮箱格式不正确"
---
02.实时数据聚合
a.聚合需求
实时统计指标：用户活跃度、交易量、转化率等
多维度聚合：按时间、地域、用户类型等维度聚合
聚合窗口选择：滑动窗口、滚动窗口、会话窗口
b.聚合架构设计
---
事件流 → 分组 → 窗口分配 → 聚合计算 → 结果输出 → 存储层
↓ ↓ ↓ ↓ ↓
分组策略 窗口类型 聚合函数 格式化 多Sink
---
c.核心代码实现
a.聚合数据模型
---
// 业务事件
public class BusinessEvent {
private String eventId;
private String eventType;
private String userId;
private String region;
private double amount;
private long timestamp;
private Map<String, Object> dimensions;
}
// 聚合结果
public class AggregationResult {
private String aggregationKey;
private long windowStart;
private long windowEnd;
private Map<String, Object> metrics;
private Map<String, String> dimensions;
private long processedCount;
}
---
b.多维聚合处理
---
// 1. 事件流接入
DataStream<BusinessEvent> eventStream = env
.addSource(new FlinkKafkaConsumer<>(
"business-events",
new BusinessEventDeserializer(),
kafkaProps
))
.name("business-events-source");
// 2. 水印生成（处理事件时间）
DataStream<BusinessEvent> withWatermarks = eventStream
.assignTimestampsAndWatermarks(
WatermarkStrategy.<BusinessEvent>forBoundedOutOfOrderness(
Duration.ofSeconds(5)
).withTimestampAssigner((event, timestamp) -> event.getTimestamp())
);
// 3. 多维度聚合
// a) 按用户聚合
KeyedStream<BusinessEvent, String> byUser = withWatermarks
.keyBy(BusinessEvent::getUserId);
DataStream<AggregationResult> userAggregates = byUser
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.aggregate(new UserAggregateFunction());
// b) 按地域聚合
KeyedStream<BusinessEvent, String> byRegion = withWatermarks
.keyBy(BusinessEvent::getRegion);
DataStream<AggregationResult> regionAggregates = byRegion
.window(SlidingEventTimeWindows.of(
Time.minutes(10), Time.minutes(1)
))
.aggregate(new RegionAggregateFunction());
// c) 按事件类型聚合
KeyedStream<BusinessEvent, String> byEventType = withWatermarks
.keyBy(BusinessEvent::getEventType);
DataStream<AggregationResult> eventTypeAggregates = byEventType
.window(SessionEventTimeWindows.withGap(Time.minutes(2)))
.aggregate(new EventTypeAggregateFunction());
// d) 复合维度聚合
DataStream<AggregationResult> compositeAggregates = withWatermarks
.keyBy(event -> event.getUserId() + "_" + event.getRegion())
.window(TumblingEventTimeWindows.of(Time.minutes(15)))
.aggregate(new CompositeAggregateFunction());
// 4. 聚合结果合并和输出
DataStream<AggregationResult> allAggregates = userAggregates
.union(regionAggregates, eventTypeAggregates, compositeAggregates);
// 5. 分级输出
allAggregates.addSink(new RedisSink(redisConfig, new RedisAggregationMapper()));
allAggregates.addSink(new ElasticsearchSink(esConfig, new ESAggregationMapper()));
allAggregates.addSink(new FlinkKafkaProducer<>(
"aggregation-results",
new AggregationResultSerializer(),
kafkaProps
));
---
c.自定义聚合函数
---
// 用户聚合函数
public class UserAggregateFunction
implements AggregateFunction<BusinessEvent, UserAccumulator, AggregationResult> {
@Override
public UserAccumulator createAccumulator() {
return new UserAccumulator();
}
@Override
public UserAccumulator add(BusinessEvent event, UserAccumulator accumulator) {
accumulator.setUserId(event.getUserId());
accumulator.setTotalAmount(accumulator.getTotalAmount() + event.getAmount());
accumulator.setEventCount(accumulator.getEventCount() + 1);
accumulator.addEventType(event.getEventType());
accumulator.updateLastActivity(event.getTimestamp());
return accumulator;
}
@Override
public AggregationResult getResult(UserAccumulator accumulator) {
AggregationResult result = new AggregationResult();
result.setAggregationKey("user_" + accumulator.getUserId());
result.setMetrics(Map.of(
"total_amount", accumulator.getTotalAmount(),
"event_count", accumulator.getEventCount(),
"avg_amount", accumulator.getTotalAmount() / accumulator.getEventCount(),
"event_types", accumulator.getEventTypes().size(),
"last_activity", accumulator.getLastActivity()
));
result.setDimensions(Map.of(
"user_id", accumulator.getUserId(),
"aggregation_type", "user"
));
return result;
}
@Override
public UserAccumulator merge(UserAccumulator a, UserAccumulator b) {
if (a.getUserId().equals(b.getUserId())) {
a.setTotalAmount(a.getTotalAmount() + b.getTotalAmount());
a.setEventCount(a.getEventCount() + b.getEventCount());
a.getEventTypes().addAll(b.getEventTypes());
a.setLastActivity(Math.max(a.getLastActivity(), b.getLastActivity()));
}
return a;
}
}
// 累加器类
public class UserAccumulator {
private String userId;
private double totalAmount = 0.0;
private int eventCount = 0;
private Set<String> eventTypes = new HashSet<>();
private long lastActivity = 0L;
// getter和setter方法
}
---
d.增量聚合实现
---
// 增量聚合函数（适用于高性能场景）
public class IncrementalAggregateFunction
extends ProcessWindowFunction<BusinessEvent, AggregationResult, String, TimeWindow> {
private ValueState<IncrementalAccumulator> accumulatorState;
@Override
public void open(Configuration parameters) {
accumulatorState = getRuntimeContext().getState(
new ValueStateDescriptor<>("incrementalAccumulator", IncrementalAccumulator.class)
);
}
@Override
public void process(String key,
Context context,
Iterable<BusinessEvent> elements,
Collector<AggregationResult> out) throws Exception {
IncrementalAccumulator accumulator = accumulatorState.value();
if (accumulator == null) {
accumulator = new IncrementalAccumulator();
}
// 增量更新
for (BusinessEvent event : elements) {
accumulator.update(event);
}
// 生成结果
AggregationResult result = accumulator.generateResult(
key, context.window()
);
out.collect(result);
// 更新状态
accumulatorState.update(accumulator);
}
}
---
e.聚合配置和优化
---
# aggregation-config.yml
aggregation:
windows:
tumbling:
- size: "1m"
lateness: "10s"
- size: "5m"
lateness: "30s"
- size: "1h"
lateness: "1m"
sliding:
- size: "10m"
slide: "1m"
lateness: "5s"
session:
- gap: "2m"
lateness: "30s"
optimization:
state_backend: "rocksdb"
checkpoint_interval: "60s"
state_ttl: "1h"
min_idle: "100ms"
max_concurrent_checkpoints: 3
---
9.4 最佳实践
01.性能优化策略
a.资源配置优化
---
# 性能调优配置
flink:
taskmanager:
memory:
process:
size: "4096m"
managed:
size: "2048m"
network:
max: "512m"
numberOfTaskSlots: 4
parallelism:
default: 4
source: 8
sink: 2
operator: 4
checkpoint:
interval: "30s"
timeout: "600s"
min_pause: "10s"
max_concurrent: 1
---
b.状态管理优化
---
// 状态TTL配置
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.hours(24))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.cleanupInRocksdbCompactFilter(1000)
.build();
ValueStateDescriptor<UserProfile> descriptor = new ValueStateDescriptor<>(
"userProfile", UserProfile.class
);
descriptor.enableTimeToLive(ttlConfig);
---
c.背压处理策略
---
// 配置背压策略
env.getConfig().setExecutionMode(ExecutionMode.PIPELINED);
env.getConfig().enableObjectReuse();
env.getConfig().setLatencyTrackingInterval(1000);
// 启用反压监控
env.getConfig().setLatencyTrackingInterval(1000);
env.getConfig().setMetricQueryInterval(30);
---
02.容错与监控
a.Checkpoint配置
---
// 高可用配置
env.enableCheckpointing(60000); // 1分钟间隔
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(600000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);
---
b.监控指标
---
// 自定义监控指标
public class MetricsProcessor extends RichMapFunction<Event, ProcessedEvent> {
private Counter processedCounter;
private Gauge<Double> processingRate;
private Histogram latencyHistogram;
@Override
public void open(Configuration parameters) {
processedCounter = getRuntimeContext()
.getMetricGroup()
.counter("processed_events");
processingRate = getRuntimeContext()
.getMetricGroup()
.gauge("processing_rate", new Gauge<Double>() {
@Override
public Double getValue() {
return calculateProcessingRate();
}
});
latencyHistogram = getRuntimeContext()
.getMetricGroup()
.histogram("event_latency");
}
@Override
public ProcessedEvent map(Event event) throws Exception {
long startTime = System.currentTimeMillis();
processedCounter.inc();
ProcessedEvent result = processEvent(event);
long latency = System.currentTimeMillis() - startTime;
latencyHistogram.update(latency);
return result;
}
}
---
c.健康检查
---
// 应用健康检查
public class FlinkHealthCheck implements HealthCheck {
private final JobID jobId;
private final JobStatusRequest jobStatusRequest;
@Override
public HealthCheckResult check() throws Exception {
try {
JobStatus jobStatus = getJobStatus(jobId);
if (jobStatus == JobStatus.RUNNING) {
return HealthCheckResult.healthy("Flink job is running");
} else {
return HealthCheckResult.unhealthy(
"Flink job status: " + jobStatus
);
}
} catch (Exception e) {
return HealthCheckResult.unhealthy(
"Failed to check Flink job status", e
);
}
}
}
---
03.部署运维
a.Docker部署
---
# Dockerfile
FROM flink:1.14.6-scala_2.11-java11
# 复制应用程序JAR
COPY target/flink-realtime-processor.jar /opt/flink/lib/
# 复制配置文件
COPY flink-conf.yaml /opt/flink/conf/
COPY log4j.properties /opt/flink/conf/
# 设置环境变量
ENV FLINK_CONF_DIR /opt/flink/conf
ENV FLINK_LIB_DIR /opt/flink/lib
# 暴露端口
EXPOSE 8081 6123
CMD ["jobmanager"]
---
b.Kubernetes部署
---
# flink-cluster.yaml
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: realtime-processor
spec:
image: flink-realtime-processor:latest
flinkVersion: v1_14
flinkConfiguration:
taskmanager.numberOfTaskSlots: "4"
state.backend: rocksdb
state.checkpoints.dir: "s3://flink-checkpoints/"
state.savepoints.dir: "s3://flink-savepoints/"
execution.checkpointing.interval: "60s"
serviceAccount: flink
jobManager:
resource:
memory: "2048m"
cpu: 2
taskManager:
resource:
memory: "4096m"
cpu: 4
replicas: 3
job:
jarURI: local:///opt/flink/lib/flink-realtime-processor.jar
parallelism: 4
upgradeMode: savepoint
state: running
---
c.运维脚本
---
#!/bin/bash
# flink-ops.sh
# 提交作业
submit_job() {
local jar_file=$1
local class_name=$2
local job_name=$3
flink run \
-d \
-c $class_name \
-p 4 \
--jobName $job_name \
-yD execution.checkpointing.interval=60s \
-yD state.backend=rocksdb \
$jar_file
}
# 保存作业状态
save_job() {
local job_id=$1
local savepoint_path=$2
flink savepoint $job_id $savepoint_path
}
# 从保存点恢复作业
restore_job() {
local jar_file=$1
local class_name=$2
local savepoint_path=$3
local parallelism=$4
flink run \
-d \
-s $savepoint_path \
-c $class_name \
-p $parallelism \
$jar_file
}
# 监控作业状态
monitor_job() {
local job_id=$1
flink list | grep $job_id
flink job $job_id
}
# 取消作业
cancel_job() {
local job_id=$1
local savepoint_path=$2
if [ -n "$savepoint_path" ]; then
flink cancel -s $savepoint_path $job_id
else
flink cancel $job_id
fi
}
---