目录
- 概述
- Flink集群搭建
- Flink运行架构
- Flink API
- Flink CEP
- 可靠性
- HA
- 监控
- 性能调优
- 参考文章
概述
- Flink是一个分布式实时和离线计算引擎，设计用于在无界和有界数据流上进行有状态的计算。它在Apache基金会旗下，作为一个开源项目被广泛开发和使用。Flink在大数据处理领域中有着重要的地位，尤其在实时数据处理方面被认为是未来的方向
- Flink的核心特性包括高吞吐、低延迟、高可用性，以及易于使用的分层API。它能在常见的集群环境中运行，并以内存速度和任意规模进行计算。Flink的另一个重要特点是它能同时处理批处理和流处理任务，使其成为一个统一的计算框架
Flink应用场景
- Flink的应用场景广泛，包括实时数据计算、实时数据仓库和ETL、事件驱动型场景（如告警、监控），以及机器学习和人工智能等。随着Flink对机器学习支持的完善，它也被用作机器学习和人工智能的处理引擎
- 实时智能推荐：根据用户历史的购买行为，通过推荐算法训练模型，预测用户未来可能会购买的物品。这可以帮助企业提升销售额，创造更大的商业价值
- 复杂事件处理：主要用于工业领域，例如对车载传感器、机械设备等实时故障检测。这些业务类型通常数据量都非常大，且对数据处理的时效性要求非常高
- 实时欺诈检测：通过实时监测交易行为，识别和预防潜在的欺诈行为。
实时数据同步、流式ETL、实时数据分析：在数据管道中处理和转换数据，包括流式数据的ETL过程，官网给了一个，这个例子很好的举例非window窗口函数用法，窗口函数时候对窗口所有数据都做处理，非窗口函数适合实时欺诈检测例子 - 窗口计算:
- 时间相关性分析：如果你需要基于时间对数据进行分析，比如统计过去一分钟内的用户点击次数、过去一小时内的交易总额等，那么窗口函数就非常适用
- 周期性报告：对于需要定期生成报告的应用，比如每天、每周或每月的统计数据，可以使用时间窗口函数来聚合这些数据
- 滑动窗口分析：滑动窗口允许你在一个固定的时间间隔内分析数据，同时窗口本身会以一定的步长滑动。这对于需要观察数据短期变化趋势的场景非常有用
- 会话分析：Flink还支持会话窗口，这种窗口类型适用于分析用户会话数据，比如用户在一次会话中的行为序列分析
- 流量控制和异常检测：通过实时统计窗口内的数据流量，可以检测到异常流量模式，从而触发警报或进行流量控制
大数据之实时流Flink总体
Flink集群搭建
Flink集群模式
- 目前线上还是推荐Yarn的per-job模式，也就是yarn模式下Flink的Application集群
- Flink Session 集群:
- 在 Flink Session 集群中，客户端连接到一个预先存在的、长期运行的集群，该集群可以接受多个作业提交。即使所有作业完成后，集群（和 JobManager）仍将继续运行直到手动停止 session 为止。因此，Flink Session 集群的寿命不受任何 Flink 作业寿命的约束
- 资源隔离：TaskManager slot 由 ResourceManager 在提交作业时分配，并在作业完成时释放。由于所有作业都共享同一集群，因此在集群资源方面存在一些竞争 — 例如提交工作阶段的网络带宽。此共享设置的局限性在于，如果 TaskManager 崩溃，则在此 TaskManager 上运行 task 的所有作业都将失败；类似的，如果 JobManager 上发生一些致命错误，它将影响集群中正在运行的所有作业
- Flink Job 集群：
- 集群生命周期：在 Flink Job 集群中，可用的集群管理器（例如 YARN）用于为每个提交的作业启动一个集群，并且该集群仅可用于该作业。在这里，客户端首先从集群管理器请求资源启动 JobManager，然后将作业提交给在这个进程中运行的 Dispatcher。然后根据作业的资源请求惰性的分配 TaskManager。一旦作业完成，Flink Job 集群将被拆除
- 资源隔离：JobManager 中的致命错误仅影响在 Flink Job 集群中运行的一个作业
- Flink Application 集群线上推荐
- Flink Application 集群是专用的 Flink 集群，仅从 Flink 应用程序执行作业，并且
main()方法在集群上而不是客户端上运行。提交作业是一个单步骤过程：无需先启动 Flink 集群，然后将作业提交到现有的 session 集群；相反，将应用程序逻辑和依赖打包成一个可执行的作业 JAR 中，并且集群入口（ApplicationClusterEntryPoint）负责调用main()方法来提取 JobGraph。例如，这允许你像在 Kubernetes 上部署任何其他应用程序一样部署 Flink 应用程序。因此，Flink Application 集群的寿命与 Flink 应用程序的寿命有关 - 在 Flink Application 集群中，ResourceManager 和 Dispatcher 作用于单个的 Flink 应用程序，相比于 Flink Session 集群，它提供了更好的隔离
Flink集群部署
部署测试实时代码
- 部署代码数据源使用模拟kafka消费的场景RandomNumberSource
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import java.util.Random;
public class RandomNumberSource extends RichParallelSourceFunction<Integer> {
private boolean isRunning = true;
@Override
public void run(SourceContext<Integer> ctx) throws Exception {
Random random = new Random();
while (isRunning) {
int randomInt = random.nextInt(10) + 1;
ctx.collect(randomInt);
Thread.sleep(5000);
}
}
@Override
public void cancel() {
isRunning = false;
}
}
- 部署代码
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
public class FlinkStreamingDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
DataStream<Integer> randomIntegers = env.addSource(new RandomNumberSource()).name("random source");
DataStream<Tuple2<Integer, Integer>> counts = randomIntegers
// map先不要用lamda不然会报错
.map(new MapFunction<Integer, Tuple2<Integer, Integer>>() {
@Override
public Tuple2<Integer, Integer> map(Integer value) throws Exception {
return new Tuple2<>(value, 1);
}
})
.assignTimestampsAndWatermarks(
WatermarkStrategy.<Tuple2<Integer, Integer>>
forMonotonousTimestamps()
.withTimestampAssigner((car, ts) -> car.f0))
.keyBy(value -> value.f0)
.window(SlidingProcessingTimeWindows.of(Time.minutes(2), Time.minutes(1)))
.sum(1);
counts.print("____________________________________");
counts.print();
env.execute("Flink Streaming Java API Skeleton V2");
}
}
部署测试离线代码
import java.util.Random;
public class OffLineDemo {
public static void main(String[] args) throws Exception {
// 创建执行环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
// 随机生成数据
String[] words = {"apple", "banana", "cherry", "durian", "elderberry", "fig", "grape"};
Random random = new Random();
DataSet<String> text = env.fromElements(
words[random.nextInt(words.length)],
words[random.nextInt(words.length)],
words[random.nextInt(words.length)],
words[random.nextInt(words.length)],
words[random.nextInt(words.length)]
);
DataSet<Tuple2<String, Integer>> counts =
// 按空格分词
text.flatMap(new Tokenizer())
// 按word字段分组
.groupBy(0)
// 对每一组的数据进行汇总
.sum(1);
// 输出结果
counts.print("+++++++++++++++++++++++++++");
counts.print();
}
public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
// emit the words
out.collect(new Tuple2<>(value, 1));
}
}
}
Flink Session模式
- Flink最全的集群部署攻略里面使用standlone模式部署
-
部署成功之后打开web ui界面，这边已经提交一个任务，就是上面实时流测试代码
web ui界面.png
running job.png
task manager.png
整体监控.png -
上传jar包之后
jar包上传.png -
配置并行度
配置并行度.png -
查看日志，查看结果
日志.png
查看结果.png -
也可取消任务，提交任务在截图的submit new job
取消任务.png - 离线代码也可以进行测试，只是离线代码类似每次提交执行一次提交执行一次，定时执行
yarn模式部署 Flink Application模式
- Flink最全的集群部署攻略里面使用arn搭建Flink集群， 前置需要看1.5万字搞定 Hadoop集群（从单机模式到分布式）的安装与部署（超详细）
-
yarn，hadoop界面
yarn，hadoop界面.png
yarn，hadoop界面.png -
yarn通过proxy url打开每个应用的flink web ui
yarn通过proxy url打开每个应用的flink web ui.png
yarn通过proxy url打开每个应用的flink web ui_1.png
yarn通过proxy url打开每个应用的flink web ui_2.png
yarn通过proxy url打开每个应用的flink web ui_3.png -
详细hadoop界面
详细hadoop界面.png
running job.png -
日志查看
日志查看.png - 命令执行
/usr/local/flink-1.16.3/bin/flink run -d -t yarn-per-job -yjm 1024 -ytm 1024 -ys 2 flink_test-1.1-SNAPSHOT-jar-with-dependencies.jar
-
web ui界面
web ui界面.png -
离线日志查看
离线日志查看.png
Flink运行架构
Flink运行架构.png
-
yarn模式per-job作业提交流程，客户端可以是linux命令提交，也可以是一些开源的web ui比如DolphinScheduler等，流程略微不太一样全程通过Yarn资源调度
yarn模式per-job作业提交流程.png
- Flink Program是我们的应用程序，当然它也可以是Flink的Web UI，或者是DolphinScheduler亦或者直接是我们通过命令行提交jar包，总之它是将应用程序提交到Yarn或者JobManager的地方
- JobManager，具有许多与协调 Flink 应用程序的分布式执行有关的职责：它决定何时调度下一个 task（或一组 task）、对完成的 task 或执行失败做出反应、协调 checkpoint、并且协调从失败中恢复等等，始终至少有一个 JobManager。高可用（HA）设置中可能有多个 JobManager，其中一个始终是 leader，其他的则是 standby
- TaskManager(也称为 worker)执行作业流的 task，并且缓存和交换数据流。必须始终至少有一个 TaskManager。在 TaskManager 中资源调度的最小单位是 task slot。TaskManager 中 task slot 的数量表示并发处理 task 的数量。请注意一个 task slot 中可以执行多个算子
JobManager
- ResourceManager: 负责 Flink 集群中的资源提供、回收、分配 - 它管理 task slots，这是 Flink 集群中资源调度的单位。Flink 为不同的环境和资源提供者(例如 YARN、Kubernetes 和 standalone 部署)实现了对应的 ResourceManager。在 standalone 设置中，ResourceManager 只能分配可用 TaskManager 的 slots，而不能自行启动新的 TaskManager
- Dispatcher: 提供了一个 REST 接口，用来提交 Flink 应用程序执行，并为每个提交的作业启动一个新的 JobMaster。它还运行 Flink WebUI 用来提供作业执行信息
- JobMaster: 负责管理单个JobGraph的执行。Flink 集群中可以同时运行多个作业，每个作业都有自己的 JobMaster
TaskManager
- 执行作业流的 task，并且缓存和交换数据流。必须始终至少有一个 TaskManager。在 TaskManager 中资源调度的最小单位是 task slot。TaskManager 中 task slot 的数量表示并发处理 task 的数量。请注意一个 task slot 中可以执行多个算子，Flink 将算子的 subtasks 链接成 tasks。每个 task 由一个线程执行。将算子链接成 task 是个有用的优化：它减少线程间切换、缓冲的开销，并且减少延迟的同时增加整体吞吐量
并行度
-
每个算子都有自己并行度，一般都说最大并行度，这里是2，整个程序包含了 7 个子任务，至少需要 2 个分区来并行执行，这段流处理程序的并行度就是 2
并行度.png - 并行度设置
- 对于一个算子，首先看在代码中是否单独指定了它的并行度，这个特定的设置优先级最高，会覆盖后面所有的设置
- 如果没有单独设置，那么采用当前代码中执行环境全局设置的并行度
- 如果代码中完全没有设置，那么采用提交时-p 参数指定的并行度
- 如果提交时也未指定-p 参数，那么采用集群配置文件中的默认并行度
- 最佳实践: 那就是在代码中只针对算子设置并行度，不设置全局并行度，这样方便我们提交作业时进行动态扩容
合并算子链
-
并行度相同的一对一(数据流维护着分区以及元素的顺序。比如图中的 source 和 map 算子，source算子读取数据之后，可以直接发送给 map 算子做处理，它们之间不需要重新分区，也不需要调整数据的顺序)算子操作，可以直接链接在一起形成一个任务
合并算子链.png - 将算子链接成 task 是非常有效的优化：可以减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量，可以手动禁止合并或者自行定义
- Flink子任务，并行度设置1时属于串行执行的情况，Flink将按照数据流中的顺序依次执行每个子任务。每个子任务处理一部分数据，并将结果传递给下一个算子或操作符。串行执行通常用于确保数据处理的顺序性和一致性。
- Flink子任务，并行度设置大于1时属于串行执行的情况，属于并行执行的情况，Flink会将一个子任务分配给多个并行实例，以便同时处理多个数据分区或数据流。并行执行可以提高数据处理的速度和效率，特别是在大规模数据处理场景中
-
保持 sink 任务并行度为 1 不变，而作业提交时设置全局并行度为 6，那么前两个任务节点就会各自有 6 个并行子任务，整个流处理程序则有 13 个子任务
Task Slot分配.png - 可以考虑把计算密集型和IO密集型的任务同时放到一个 slot 中，它们就可以自行分配对资源占用的比例，从而保证最重的活平均分配给所有的
也可以通过设置slot 共享组（SlotSharingGroup）手动指定，只有属于同一个 slot 共享组的子任务，才会开启 slot 共享
sink算子
- 写入文件:
writeAsText(...): 将数据以文本格式写入文件
writeAsCsv(...): 将数据以 CSV 格式写入文件 - 打印输出:
print(): 在标准输出上打印数据流的内容 - 写入 Socket:
writeToSocket(...): 将数据写入 TCP 套接字 - 消息队列和流处理平台:
Apache Kafka: 使用 Flink Kafka Connector 将数据写入 Kafka 主题
RabbitMQ: 使用 Flink RabbitMQ Connector 将数据写入 RocketMQ 队列 - 数据库存储:
JDBC: 使用 Flink JDBC Connector 将数据写入关系型数据库
Elasticsearch: 使用 Flink Elasticsearch Connector 将数据写入 Elasticsearch 索引
Cassandra: 使用 Flink Cassandra Connector 将数据写入 Apache Cassandra 数据库
Redis: 使用 Flink Redis Connector 将数据写入 Redis 数据库 - 其他数据存储:
Apache Hadoop FileSystem (HDFS): 将数据写入 HDFS 文件系统。
自定义 Sink: 实现 RichSinkFunction 或 SinkFunction 接口，将数据写入自定义的数据存储系统 - 这些 source 和 sink 只是 Flink 支持的一部分，Flink 社区还在不断扩展和增加新的 connector 以支持更多的数据源和数据存储系统。您可以根据自己的需求选择合适的 source 和 sink 进行数据处理和传输
各种中间操作转换算子
时间窗口函数
- Map: 用途是对每个元素执行指定的转换操作, eg将每个字符串转换为大写
- FlatMap: 用途是将一个元素转换为一个零个、一个或多个输出元素, eg将每个单词拆分为单个字符
- Filter: 用途是根据条件筛选数据流中的元素, eg过滤掉长度小于 5 的单词
- KeyBy: 用途是根据指定的键将数据流中的元素进行分组, eg按单词的第一个字母对单词进行分组。
- Reduce: 用途是对数据流中的元素进行聚合操作, eg计算单词的长度总和。
- Join: 用途是将两个数据流连接在一起，基于指定的键进行连接操作, eg将用户点击流与广告数据流连接，基于用户ID进行连接。
- CoGroup: 用途是对两个数据流进行分组，并执行类似于 MapReduce 的操作, eg根据用户ID对用户点击流和广告数据流进行分组，并计算每个用户点击的广告数量
- Union: 用途是将两个数据流合并为一个数据流, eg将两个单词流合并为一个更大的单词流
- Window: 用途是将数据流划分为时间窗口或计数窗口，并在窗口上执行聚合操作, eg计算每个小时内单词的数量总和
- BroadcastStream:
用途是将一个不变的数据流广播到每个并行实例上，通常用于状态管理和并行算法中
- 举个简单例子keyby + 时间窗口 + process函数
- 更多例子可参考官网和业务实战场景(十一)实时流Flink实战
Flink API
source算子
- Flink支持多种类型的Source，包括基于本地集合的Source、基于文件的Source、基于网络套接字的Source和自定义的Source。以下是对这些Source的简单介绍：
- 基于本地集合的Source：这种Source允许用户通过env.fromElements()方法创建一个本地集合的数据集，支持Tuple、自定义对象等多种复合形式
- 基于文件的Source：这种Source用于从文件中读取数据，包括readTextFile、socketTextStream和readFile等方法。这些方法可以读取文本文件、CSV文件等格式的文件，并转换为Flink的数据集
- 基于网络套接字的Source：这种Source可以从网络套接字中读取数据，通常用于从实时数据流中读取数据。Flink提供了基于TCP、UDP等协议的Source，可以根据实际需求选择使用
- 自定义的Source：如果以上类型的Source无法满足需求，Flink还允许用户自定义Source。用户需要实现SourceFunction接口，并将该实现类的实例作为参数传入到StreamExecutionEnvironment的addSource方法中
- 除了以上类型的Source，Flink还支持许多第三方Connector Source，例如Apache Kafka、Amazon Kinesis、RocketMQ、Twitter Streaming API、Apache NiFi等。这些Connector Source使得Flink可以轻松地从各种数据源中读取数据，并进行实时流处理， 举例
DataStream<Tuple2<String, Integer>> input = env.fromElements(
new Tuple2<>("key1", 1),
new Tuple2<>("key1", 2),
new Tuple2<>("key1", 3),
new Tuple2<>("key2", 10),
new Tuple2<>("key2", 11),
new Tuple2<>("key2", 12)
);
// 使用KeyBy进行分组，并使用TimeWindow进行滑动窗口处理
input
.keyBy(0) // 使用第一个元素作为键进行分组
.timeWindow(Time.seconds(1), Time.seconds(1)) // 定义滑动窗口大小为1秒，窗口间隔也为1秒
.process(new ProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
@Override
public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
// 在这里处理每个窗口内的数据，例如计算窗口内的总和
int sum = ctx.getWindow().getEnd() - ctx.getWindow().getStart(); // 计算窗口内的元素数量
out.collect(new Tuple2<>(value.f0, sum)); // 输出每个窗口内的总和和对应的键
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
// 处理定时器触发的事件，例如清理旧窗口的数据等操作
}
});
Flink CEP
- 主要功能：
- 事件流处理：Flink CEP能够处理高速、大规模的事件流数据，包括各种来源和格式的事件
- 模式检测：Flink CEP使用CEP算法，能够在事件流中发现符合特定模式的序列。这些模式可以定义为一系列相关事件的组合，例如股票交易中的特定价格变动序列
- 实时处理：Flink CEP支持实时事件流处理，能够及时发现符合特定模式的序列，并提供实时的反馈或警告
- 灵活的规则定义：Flink CEP提供了灵活的API，允许用户通过编程方式定义复杂的模式匹配规则。这些规则可以基于事件的内容、顺序、频率和时间间隔等
- 高效性能：Flink CEP利用Flink的分布式处理能力，能够高效地处理大规模事件流数据，并提供高吞吐量和低延迟的输出结果
- 集成与扩展：Flink CEP可以与其他Flink组件集成，如SQL API、Table API等，并支持与其他系统的集成。此外，Flink CEP还提供了扩展点，允许用户自定义或集成其他算法和工具
7 可视化监控与调试：Flink CEP提供了可视化工具，帮助用户监控事件流的处理过程、模式匹配结果等，方便调试和优化处理逻辑 - 可靠性：Flink CEP支持容错机制，确保在系统故障或数据丢失的情况下能够恢复到一致的状态
- 可扩展性：随着业务需求的增长和变化，Flink CEP提供了灵活的扩展能力，支持自定义模式匹配算法、事件格式等
- 例子可以看官网FlinkCEP - Flink的复杂事件处理
可靠性
- Flink通过一系列机制保证exactly-once（精确一次）的可靠性。这些机制主要包括：
- 检查点（Checkpoint）：Flink使用检查点来确保在处理故障时能够恢复到正确的状态。检查点允许Flink在特定时间点上保存应用程序的状态，以便在发生故障时进行恢复。当应用程序从一个检查点恢复时，它可以确保从该检查点开始的所有数据都被正确处理，从而实现exactly-once的语义
- 两阶段提交协议（Two-Phase Commit Protocol）：Flink使用两阶段提交协议来协调检查点和数据输出。在第一阶段（pre-commit），Flink将处理结果写入到外部存储系统，但不立即提交。只有当检查点成功完成时，Flink才会进入第二阶段（commit），并提交之前写入的数据。如果发生故障，Flink将回滚到上一个成功的检查点，并重新处理该检查点之后的数据，从而确保数据的一致性，其过程分为以下几个步骤：
2.1 开始事务（beginTransaction）：创建一个临时文件夹，用于将数据写入其中
2.2 预提交（preCommit）：将内存中缓存的数据写入文件并关闭。此时，数据已经写入到临时文件夹中，但还未提交到目标目录
2.3 正式提交（commit）：将之前写完的临时文件放入目标目录下。这代表着最终的数据会有一些延迟，但是数据的一致性得到了保证
2.4 丢弃（abort）：若出现故障或异常，可以选择丢弃临时文件 - 可重放的数据源（Replayable Data Sources）：为了实现exactly-once的语义，数据源需要支持重放功能。这意味着在发生故障并恢复后，数据源能够重新提供与之前相同的数据。Flink通过与数据源进行协调，确保在恢复时能够从正确的位置开始重新读取数据
- 幂等性写入和事务性写入（Idempotent and Transactional Writes）：对于输出到外部系统的数据，Flink需要确保在发生故障时能够安全地重试写入操作，而不会导致数据重复或丢失。这通常通过幂等性写入或事务性写入来实现。幂等性写入意味着多次执行相同的写入操作将产生相同的结果，而事务性写入则使用事务机制来确保数据的一致性和完整性
HA
- 基于zk实现高可用ZooKeeper 高可用服务
- standlone部署时配置只需配置Yarn HA：首先确保Yarn集群配置了高可用性（HA）。这包括配置ResourceManager和NodeManager的高可用性。确保Yarn的HA配置正确设置，包括Zookeeper作为协调服务
- 配置Flink HA：为了实现Flink的高可用性，需要配置Flink的HA模式。这通常涉及使用Checkpointing机制来确保JobManager状态的一致性。确保Flink的HA模式正确配置，并能够利用Yarn的HA机制
监控
- Flink提供了多种监控方式，包括系统级别的监控、JVM级别的监控以及Flink job/Task/Operator级别的监控
- 系统级别的监控包括CPU状态信息、内存状态信息等，可以通过更改配置文件打开，同时需要添加相关依赖jar包。JVM级别的监控则关注CPU占用率、堆内存使用情况、线程数量、GC已经类加载数量等
- Flink job/Task/Operator级别的监控则更加具体，包括检查点相关（耗时、大小、数量）、算子吞吐量、水位线等。对于使用FlinkKafkaConsumer的用户，可以监控job消费kafka速度、offset的LAG值。这些级别的监控也支持自定义
- 另外，Flink也提供了Rest API来获取监控数据，通过Chrome浏览器的控制台可以查看这些API。这些API能够获取到很多重要的监控信息，如算子的吞吐量、水位线等。然后可以告警
- 除此之外，也可以通过Prometheus和Grafana等工具来对Flink进行监控。具体来说，可以使用PushGateway来推送指标，Prometheus来采集及存储指标，Grafana来配置Dashboard，让指标进行可视化。然后可以定义告警
- 总的来说，Flink的监控体系是多层次、全方位的，既有自带的监控工具和API，也有开源工具可以选择使用，从而帮助用户更好地了解Flink的运行状态并进行相应的优化
性能调优
参考文章
- 1.5万字搞定 Hadoop集群（从单机模式到分布式）的安装与部署（超详细）
- Flink最全的集群部署攻略（推荐yarn实现企业级部署）
- Apache Flink 文档
- Apache Flink On Yarn模式高可用(HA)集群部署
- 一文详解Flink on Yarn的三种部署方式及使用说明
- Flink on yarn以及实现jobManager 高可用(HA)
- Flink on Yarn集群HA高可用部署 基于flink1.12 hadoop 3.0 CDH6.3.2
- Flink on Yarn模式部署
- Flink classloader.check-leaked-classloader报错
- flink 提交程序报错No Executor found. Please make sure to export the HADOOP_CLASSPATH environment variable
- Flink 1.17教程：部署模式介绍及Standalone运行模式
- Apache Flink 零基础入门（三）：开发环境搭建和应用的配置、部署及运行
- Spark on YARN 部署搭建详细图文教程
- Flink CEP
