一、Flink 架构
Flink 架构 | Apache Flink
二、设置TaskManager、Slot和Parallelism
在Apache Flink中，设置TaskManager、Slot和Parallelism是配置Flink集群性能和资源利用的关键步骤。以下是关于如何设置这些参数的详细指南：
1. TaskManager 设置
TaskManager是Flink集群中负责执行作业的节点。关于TaskManager的设置，主要关注其数量和资源分配。
- TaskManager数量：根据集群规模和作业需求确定TaskManager的数量。例如，如果集群资源充足且作业并发度高，可以增加TaskManager的数量以提高处理能力。
- 资源分配：为每个TaskManager分配适当的内存和CPU资源。这取决于集群的硬件配置和作业的资源需求。确保为TaskManager分配足够的资源以确保作业可以高效运行。
2. Slot 设置
Slot是TaskManager上用于执行作业的资源单元。一个Slot可以并行运行一个作业的子任务。
- Slot数量：每个TaskManager上的Slot数量决定了该TaskManager可以并行运行的作业子任务数。Slot数量通常根据TaskManager的内存和CPU资源来确定。例如，如果TaskManager有2GB内存和1个CPU核心，并且每个Slot需要1GB内存和0.5个CPU核心，则该TaskManager可以设置2个Slot。
- 资源分配：每个Slot会分配到一定的内存和CPU资源。这些资源应该根据作业的需求和TaskManager的总资源进行合理分配。
3. Parallelism 设置
Parallelism决定了Flink作业的并行度，即作业可以并行执行的程度。
- 默认并行度：在Flink配置文件中，可以指定默认并行度（
parallelism.default）。如果作业没有指定并行度，则使用默认并行度。 - 作业级并行度：在提交作业时，可以通过命令行参数（
-p）或编程API（env.setParallelism()）为整个作业设置并行度。这将作为作业的默认并行度，但可以被单个算子的并行度设置覆盖。 - 算子级并行度：在Flink程序中，可以为每个算子单独设置并行度。这可以通过在算子链的末尾调用
setParallelism()方法来实现。算子级并行度的优先级高于作业级并行度和默认并行度。
4. 总结
- 设置TaskManager的数量和资源分配以适应集群规模和作业需求。
- 根据TaskManager的资源为每个TaskManager设置适当的Slot数量。
- 根据作业的需求和集群的资源设置作业的默认并行度、作业级并行度和算子级并行度。
5. 阿里云 实时计算Flink版 参数示例
三、Flink SQL性能调优与配置
在使用Flink SQL进行数据处理时，性能调优是确保系统高效运行的关键。以下是一些常见的调优配置和策略，它们可以帮助您优化Flink SQL作业的性能。
1. 微批处理（Mini-Batch）
Flink SQL支持微批处理，通过组合多个小批次来减少任务调度的开销。当启用微批处理时，Flink会尝试将多个小批次合并成一个较大的批次进行处理。
# 启用微批处理 |
table.exec.mini-batch.enabled: 'true' |
# 设置允许的最大延迟时间，超过该时间将不再等待更多数据而直接发送当前批次 |
table.exec.mini-batch.allow-latency: 2s |
2. 算子链优化（Operator Chaining）
算子链优化是一种减少任务间数据传输开销的策略。通过将多个算子链接在一起，可以减少序列化和反序列化的开销，并提高数据传输的效率。
# 默认情况下，Flink会尝试自动进行算子链优化 |
# 如果需要禁用此功能，可以设置为false |
pipeline.operator-chaining: 'false' |
注意：通常建议保持算子链优化开启（'true'），以获得更好的性能。
3. Hash Shuffle
在Flink中，Keyed Streams使用hash shuffle策略将数据分发到下游的并行任务。这有助于确保具有相同key的数据被发送到同一个下游任务，从而进行高效的聚合或连接操作。
对于Flink SQL中的sink，如果其接受的是Keyed Stream，并且需要确保数据的顺序性，可以使用FORCE关键字来强制使用hash shuffle。
# 强制使用hash shuffle |
table.exec.sink.keyed-shuffle: FORCE |
注意：在Flink SQL中，您通常不需要手动配置这个参数，因为Flink会根据作业的特性和需求自动选择合适的shuffle策略。
4. Hash Join
Hash Join是一种基于哈希表的连接算法，适用于等值连接场景。它通过将一个表的数据加载到哈希表中，然后扫描另一个表并与哈希表中的数据进行比较来实现连接。
在Flink SQL中，可以使用Hint（提示）来建议优化器使用Hash Join。但是，请注意，这只是一个建议，优化器可能会根据实际情况选择其他连接策略。
SELECT /*+ SHUFFLE_HASH(t1,t2) */ * FROM table1 t1 JOIN table2 t2 ON t1.key = t2.key |
5. 设置Sink的并行度
Sink的并行度决定了数据写入外部系统时的并行度。可以根据外部系统的性能和Flink作业的需求来设置合适的并行度。
在Flink SQL中，可以通过DDL语句或API来设置Sink的并行度。以下是一个示例DDL语句：
CREATE TABLE sink_table ( |
... -- 定义表结构 |
) WITH ( |
... -- 其他配置选项 |
'sink.parallelism' = '4' -- 设置并行度为4 |
); |
或者，在Flink作业提交时通过API来动态设置Sink的并行度。
