【Flink】Flink SQL

一、Flink 架构

Flink 架构 | Apache Flink

二、设置TaskManager、Slot和Parallelism

在Apache Flink中，设置TaskManager、Slot和Parallelism是配置Flink集群性能和资源利用的关键步骤。以下是关于如何设置这些参数的详细指南：

**1. **TaskManager 设置

TaskManager是Flink集群中负责执行作业的节点。关于TaskManager的设置，主要关注其数量和资源分配。

1. TaskManager数量：根据集群规模和作业需求确定TaskManager的数量。例如，如果集群资源充足且作业并发度高，可以增加TaskManager的数量以提高处理能力。
2. 资源分配：为每个TaskManager分配适当的内存和CPU资源。这取决于集群的硬件配置和作业的资源需求。确保为TaskManager分配足够的资源以确保作业可以高效运行。

**2. **Slot 设置

Slot是TaskManager上用于执行作业的资源单元。一个Slot可以并行运行一个作业的子任务。

1. Slot数量：每个TaskManager上的Slot数量决定了该TaskManager可以并行运行的作业子任务数。Slot数量通常根据TaskManager的内存和CPU资源来确定。例如，如果TaskManager有2GB内存和1个CPU核心，并且每个Slot需要1GB内存和0.5个CPU核心，则该TaskManager可以设置2个Slot。
2. 资源分配：每个Slot会分配到一定的内存和CPU资源。这些资源应该根据作业的需求和TaskManager的总资源进行合理分配。

**3. **Parallelism 设置

Parallelism决定了Flink作业的并行度，即作业可以并行执行的程度。

1. 默认并行度：在Flink配置文件中，可以指定默认并行度（parallelism.default）。如果作业没有指定并行度，则使用默认并行度。
2. 作业级并行度：在提交作业时，可以通过命令行参数（-p）或编程API（env.setParallelism()）为整个作业设置并行度。这将作为作业的默认并行度，但可以被单个算子的并行度设置覆盖。
3. 算子级并行度：在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的并行度。