自学内容网 自学内容网
自学内容网 > 正文

Flink调优详解：案例解析（第42天）

🕗 发布于 2024-07-21 09:15 flink  大数据  

系列文章目录

一、Flink-任务参数配置
二、Flink-SQL调优
三、阿里云Flink调优

文章目录

前言

本文主要详解常见的Flink优化策略。

官网地址：https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/config/

一、Flink-任务参数配置

1.1 运行时参数
  • 异步维度join
# 默认值：100
# 值类型：Integer
# 流批任务：流、批任务都支持
# 用处：异步 lookup join 中最大的异步 IO 执行数目
table.exec.async-lookup.buffer-capacity: 100
  • 开启微批
# 默认值：false
# 值类型：Boolean
# 流批任务：流任务支持
# 用处：MiniBatch 优化是一种专门针对 unbounded 流任务的优化（即非窗口类应用），其机制是在 `允许的延迟时间间隔内` 以及 `达到最大缓冲记录数` 时触发以减少 `状态访问` 的优化，从而节约处理时间。下面两个参数一个代表 `允许的延迟时间间隔`，另一个代表 `达到最大缓冲记录数`。
table.exec.mini-batch.enabled: false

# 默认值：0 ms
# 值类型：Duration
# 流批任务：流任务支持
# 用处：此参数设置为多少就代表 MiniBatch 机制最大允许的延迟时间。注意这个参数要配合 `table.exec.mini-batch.enabled` 为 true 时使用，而且必须大于 0 ms
table.exec.mini-batch.allow-latency: 0 ms

# 默认值：-1
# 值类型：Long
# 流批任务：流任务支持
# 用处：此参数设置为多少就代表 MiniBatch 机制最大缓冲记录数。注意这个参数要配合 `table.exec.mini-batch.enabled` 为 true 时使用，而且必须大于 0
table.exec.mini-batch.size: -1


注意: 
1- 如果想开启微批，那么首先需要将第一个参数值设置为true；接着可以设置后面两个参数。只要程序满足后面2个参数中的任意一个条件就会触发微批处理。
2- 微批处理能够提升程序处理数据的吞吐量，但是会导致数据的处理时效性降低
  • 并行度的设置
# 默认值：-1
# 值类型：Integer
# 流批任务：流、批任务都支持
# 用处：可以用此参数设置 Flink SQL 中算子的并行度，这个参数的优先级 `高于` StreamExecutionEnvironment 中设置的并行度优先级，如果这个值设置为 -1，则代表没有设置，会默认使用 StreamExecutionEnvironment 设置的并行度
table.exec.resource.default-parallelism: -1
  • 数据异常时的处理方式
# 默认值：ERROR
# 值类型：Enum【ERROR, DROP】
# 流批任务：流、批任务都支持
# 用处：表上的 NOT NULL 列约束强制不能将 NULL 值插入表中。Flink 支持 `ERROR`（默认）和 `DROP` 配置。默认情况下，当 NULL 值写入 NOT NULL 列时，Flink 会产生运行时异常。用户可以将行为更改为 `DROP`，直接删除此类记录，而不会引发异常。
table.exec.sink.not-null-enforcer: ERROR
  • 上游cdc去重
# 默认值：false
# 值类型：Boolean
# 流批任务：流任务
# 用处：接入了 CDC 的数据源，上游 CDC 如果产生重复的数据，可以使用此参数在 Flink 数据源算子进行去重操作，去重会引入状态开销
table.exec.source.cdc-events-duplicate: false
  • 设置空闲等待
# 默认值：0 ms
# 值类型：Duration
# 流批任务：流任务
# 用处：如果此参数设置为 60 s，当 Source 算子在 60 s 内未收到任何元素时，这个 Source 将被标记为临时空闲，此时下游任务就不依赖此 Source 的 Watermark 来推进整体的 Watermark 了。
# 默认值为 0 时，代表未启用检测源空闲。
table.exec.source.idle-timeout: 0 ms
  • 设置状态有效期
# 默认值：0 ms
# 值类型：Duration
# 流批任务：流任务
# 用处：指定空闲状态（即未更新的状态）将保留多长时间。尤其是在 unbounded 场景中很有用。默认 0 ms 为不清除空闲状态
table.exec.state.ttl: 0 ms

推荐: ttl time to live失效时间。保留窗口附近的1-3个state状态数据。举例: 例如滚动窗口大小是5秒，那么这个参数推荐设置为5000 - 15000毫秒之间

上述的参数中，常用的有：开启微批、设置状态有效期

1.2 优化器参数
  • 开启两阶段聚合
#  默认值：AUTO
#  值类型：String
#  流批任务：流、批任务都支持
#  用处：聚合阶段的策略。和 MapReduce 的 Combiner 功能类似，可以在数据 shuffle 前做一些提前的聚合，可以选择以下三种方式
#  TWO_PHASE：强制使用具有 localAggregate 和 globalAggregate 的两阶段聚合。请注意，如果聚合函数不支持优化为两个阶段，Flink 仍将使用单阶段聚合。
#  两阶段优化在计算 count，sum 时很有用，但是在计算 count distinct 时需要注意，key 的稀疏程度，如果 key 不稀疏，那么很可能两阶段优化的效果会适得其反
#  ONE_PHASE：强制使用只有 CompleteGlobalAggregate 的一个阶段聚合。
#  AUTO：聚合阶段没有特殊的执行器。选择 TWO_PHASE 或者 ONE_PHASE 取决于优化器的成本。
table.optimizer.agg-phase-strategy: AUTO

注意: 两阶段聚合必须配合微批处理的参数一起进行设置。
  • 开启分桶
#  默认值：false
#  值类型：Boolean
#  流批任务：流任务
#  用处：避免 group by 计算 count distinct\sum distinct 数据时的 group by 的 key 较少导致的数据倾斜，比如 group by 中一个 key 的 distinct 要去重 500w 数据，而另一个 key 只需要去重 3 个 key，那么就需要先需要按照 distinct 的 key 进行分桶。将此参数设置为 true 之后，下面的 table.optimizer.distinct-agg.split.bucket-num 可以用于决定分桶数是多少
#  后文会介绍具体的案例
table.optimizer.distinct-agg.split.enabled: false

#  默认值：1024
#  值类型：Integer
#  流批任务：流任务
#  用处：避免 group by 计算 count distinct 数据时的 group by 较少导致的数据倾斜。加了此参数之后，会先根据 group by key 结合 hash_code（distinct_key）进行分桶，然后再自动进行合桶。
#  后文会介绍具体的案例
table.optimizer.distinct-agg.split.bucket-num: 1024
  • 重用执行计划
#  默认值：true
#  值类型：Boolean
#  流批任务：流任务
#  用处：如果设置为 true，Flink 优化器将会尝试找出重复的自计划并重用。默认为 true 不需要改动
table.optimizer.reuse-sub-plan-enabled: true
  • source资源重用
#  默认值：true
#  值类型：Boolean
#  流批任务：流任务
#  用处：如果设置为 true，Flink 优化器会找出重复使用的 table source 并且重用。默认为 true 不需要改动
table.optimizer.reuse-source-enabled: true
  • 开启谓词下推
#  默认值：true
#  值类型：Boolean
#  流批任务：流任务
#  用处：如果设置为 true，Flink 优化器将会做谓词下推到 FilterableTableSource 中，将一些过滤条件前置，提升性能。默认为 true 不需要改动
table.optimizer.source.predicate-pushdown-enabled: true
1.3 表参数
  • 开启DML异步
#  默认值：false
#  值类型：Boolean
#  流批任务：流、批任务都支持
#  用处：DML SQL（即执行 insert into 操作）是异步执行还是同步执行。默认为异步（false），即可以同时提交多个 DML SQL 作业，如果设置为 true，则为同步，第二个 DML 将会等待第一个 DML 操作执行结束之后再执行
table.dml-sync: false
  • 设置方法的最大长度不超过64KB
#  默认值：64000
#  值类型：Integer
#  流批任务：流、批任务都支持
#  用处：Flink SQL 会通过生产 java 代码来执行具体的 SQL 逻辑，但是 jvm 限制了一个 java 方法的最大长度不能超过 64KB，但是某些场景下 Flink SQL 生产的 java 代码会超过 64KB，这时 jvm 就会直接报错。因此此参数可以用于限制生产的 java 代码的长度来避免超过 64KB，从而避免 jvm 报错。
table.generated-code.max-length: 64000
  • 本地时区
#  默认值：default
#  值类型：String
#  流批任务：流、批任务都支持
#  用处：在使用天级别的窗口时，通常会遇到时区问题。举个例子，Flink 开一天的窗口，默认是按照 UTC 零时区进行划分，那么在北京时区划分出来的一天的窗口是第一天的早上 8:00 到第二天的早上 8:00，但是实际场景中想要的效果是第一天的早上 0:00 到第二天的早上 0:00 点。因此可以将此参数设置为 GMT+08:00 来解决这个问题。
table.local-time-zone: default
  • 编译器
#  默认值：default
#  值类型：Enum【BLINK、OLD】
#  流批任务：流、批任务都支持
#  用处：Flink SQL planner，默认为 BLINK planner，也可以选择 old planner，但是推荐使用 BLINK planner
table.planner: BLINK
  • SQL方言
#  默认值：default
#  值类型：String
#  流批任务：流、批任务都支持
#  用处：Flink 解析一个 SQL 的解析器，目前有 Flink SQL 默认的解析器和 Hive SQL 解析器，其区别在于两种解析器支持的语法会有不同，比如 Hive SQL 解析器支持 between and、rlike 语法，Flink SQL 不支持
table.sql-dialect: default

二、Flink-SQL调优

2.1 mini-batch聚合

作用：微批处理能够提升程序处理数据的吞吐量，但是会导致数据的处理时效性降低

在这里插入图片描述

SQL中参数配置如下：

# 默认值：false
# 值类型：Boolean
# 流批任务：流任务支持
# 用处：MiniBatch 优化是一种专门针对 unbounded 流任务的优化（即非窗口类应用），其机制是在 `允许的延迟时间间隔内` 以及 `达到最大缓冲记录数` 时触发以减少 `状态访问` 的优化，从而节约处理时间。下面两个参数一个代表 `允许的延迟时间间隔`，另一个代表 `达到最大缓冲记录数`。
table.exec.mini-batch.enabled: false

# 默认值：0 ms
# 值类型：Duration
# 流批任务：流任务支持
# 用处：此参数设置为多少就代表 MiniBatch 机制最大允许的延迟时间。注意这个参数要配合 `table.exec.mini-batch.enabled` 为 true 时使用，而且必须大于 0 ms
table.exec.mini-batch.allow-latency: 0 ms

# 默认值：-1
# 值类型：Long
# 流批任务：流任务支持
# 用处：此参数设置为多少就代表 MiniBatch 机制最大缓冲记录数。注意这个参数要配合 `table.exec.mini-batch.enabled` 为 true 时使用，而且必须大于 0
table.exec.mini-batch.size: -1
2.2 两阶段聚合

作用：1- 提前对数据进行局部聚合操作，能够减少后续数据处理量；2-能够一定程度避免数据倾斜的问题

在这里插入图片描述

FlinkSQL中的配置：

#  默认值：AUTO
#  值类型：String
#  流批任务：流、批任务都支持
#  用处：聚合阶段的策略。和 MapReduce 的 Combiner 功能类似，可以在数据 shuffle 前做一些提前的聚合，可以选择以下三种方式
#  TWO_PHASE：强制使用具有 localAggregate 和 globalAggregate 的两阶段聚合。请注意，如果聚合函数不支持优化为两个阶段，Flink 仍将使用单阶段聚合。
#  两阶段优化在计算 count，sum 时很有用，但是在计算 count distinct 时需要注意，key 的稀疏程度，如果 key 不稀疏，那么很可能两阶段优化的效果会适得其反
#  ONE_PHASE：强制使用只有 CompleteGlobalAggregate 的一个阶段聚合。
#  AUTO：聚合阶段没有特殊的执行器。选择 TWO_PHASE 或者 ONE_PHASE 取决于优化器的成本。
table.optimizer.agg-phase-strategy: AUTO


注意: 如果想要使用两阶段聚合，那么必须同时开始mini-batch微批处理
2.3 分桶

作用：能够一定程度避免数据倾斜的问题

在这里插入图片描述

FlinkSQL的配置如下：

#  默认值：false
#  值类型：Boolean
#  流批任务：流任务
#  用处：避免 group by 计算 count distinct\sum distinct 数据时的 group by 的 key 较少导致的数据倾斜，比如 group by 中一个 key 的 distinct 要去重 500w 数据，而另一个 key 只需要去重 3 个 key，那么就需要先需要按照 distinct 的 key 进行分桶。将此参数设置为 true 之后，下面的 table.optimizer.distinct-agg.split.bucket-num 可以用于决定分桶数是多少
#  后文会介绍具体的案例
table.optimizer.distinct-agg.split.enabled: false

#  默认值：1024
#  值类型：Integer
#  流批任务：流任务
#  用处：避免 group by 计算 count distinct 数据时的 group by 较少导致的数据倾斜。加了此参数之后，会先根据 group by key 结合 hash_code（distinct_key）进行分桶，然后再自动进行合桶。
#  后文会介绍具体的案例
table.optimizer.distinct-agg.split.bucket-num: 1024
2.4 filter去重（了解）
--普通的写法
SELECT
 day,
 COUNT(DISTINCT user_id) AS total_uv,
 COUNT(DISTINCT CASE WHEN flag IN ('android', 'iphone') THEN user_id ELSE NULL END) AS app_uv,
 COUNT(DISTINCT CASE WHEN flag IN ('wap', 'other') THEN user_id ELSE NULL END) AS web_uv
FROM T
GROUP BY day


--filter优化写法
SELECT
 day,
 COUNT(DISTINCT user_id) AS total_uv,
 COUNT(DISTINCT user_id) FILTER (WHERE flag IN ('android', 'iphone')) AS app_uv,
 COUNT(DISTINCT user_id) FILTER (WHERE flag IN ('web', 'other')) AS web_uv
FROM T
GROUP BY day

filter子句能够将三个状态合并成一个共享的状态。方便程序的读取等操作。能够提升效率。

在这里插入图片描述

三、阿里云Flink调优

Flink支持智能调优和定时调优两种调优模式。

在这里插入图片描述

3.1 智能调优

简单理解，就是阿里云Flink智能来进行调优。默认会从并发度和内存来进行调优。

  • 智能调优会调整作业的并发度来满足作业流量变化所需要的吞吐。

    智能调优会监控消费源头数据的延迟变化情况、TaskManager（TM） CPU实际使用率和各个算子处理数据能力来调整作业的并发度。详情如下：

    • 作业延迟Delay指标正常（不超过60s），不修改当前作业并发。
    • 作业延迟Delay指标超过默认阈值60s，分以下两种情况来调整并发度：
      • 延迟正在下降，不进行并发度调整。
      • 延迟增加并且连续上升3分钟（默认值）， 默认调整作业并发度到当前实际TPS的两倍，但不超过设置最大的资源（默认值为64 CU）。
    • 作业不存在延迟指标。
      • 作业某VERTEX节点连续6分钟实际处理数据时间占比超过80%，调大作业并发度使得SLOT使用率降低到50%，但不超过设置最大的资源（默认为64 CU）。
      • 所有TM的平均利用率连续6分钟超过80%，调高并发度使TM的CPU使用率降低到50%。
    • 所有TM的最大CPU使用率连续24小时低于20%，且VERTEX的实际处理数据时间低于20%时，调低作业的并发度使CPU和VERTEX实际处理的时间占比提高到50%。

  • 智能调优也会监控作业的内存使用和Failover情况，来调整作业的内存配置。详情如下：

    • 在JobManager GC频繁或者发生OOM异常时，会调高JM的内存，默认最大调整到16 GiB。
    • 在TM GC频繁或者发生OOM异常、HeartBeatTimeout异常时，会调高TM的内存，默认最大调整到16 GiB。
    • 在TM内存使用率超过95%时，会调大TM的内存。
    • 在TM的实际内存使用率连续24小时低于30%时，降低TM内存的配置，默认最小调整到1.6 GiB。
3.2 定时调优

定时调优计划描述了资源和时间点的对应关系，一个定时调优计划中可以包含多组资源和时间点的关系。在使用定时调优计划时，您需要明确知道各个时间段的资源使用情况，根据业务时间区间特征，设置对应的资源。

例如，某业务全天早09：00~19：00是业务高峰，19：00到第二天09：00是业务低峰。此时您可以使用定时调优功能，在高峰时间段使用30 CU，在业务低峰时使用10 CU。

说明：1 CU=1核CPU+4 GiB内存+20 GB本地存储（放置日志、系统检查点等信息）

小结：在使用阿里云调优时，可以使用一些开源的参数。


原文地址：https://blog.csdn.net/syhiiu/article/details/140542743

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

相关文章

  • MinIo在Ubantu和Java中的整合

    方法抛出的各种异常进行了捕获，然后打印了异常信息，目前这种处理逻辑，无论Minio是否发生异常，前端在上传文件时，总是会受到成功的响应信息。会处理所有Controller方法抛出的异常，因此Contr

    阅读更多2024-11-15
  • HTTP基础

    当浏览者访问一个网页时，浏览者的浏览器会向网页所在服务器发出请求。当浏览器接收并显示网页前，此网页所在的服务器会返回一个包含HTTP状态码的信息头（server header）用以响应浏览器的请求。H

    阅读更多2024-11-15
  • linux phy mdio 读取工具

    【代码】linux phy mdio 读取工具。

    阅读更多2024-11-15
  • sql文件

    sql文件通常包含SQL语句，用于数据库的创建、修改和数据操作。根据内容的不同，.sql文件的使用方式也有所不同。

    阅读更多2024-11-15
  • MongoDB创建只读用户并授权指定集合的查询权限

    创建一个自定义角色，只允许在。集合上执行查询操作。

    阅读更多2024-11-15
  • #渗透测试#SRC漏洞挖掘#云技术基础03之容器相关

    Podman是Docker的替代产品，它无守护进程。在运行容器时，若不加sudo启动可能会报错，因为默认禁止侦听1024以下端口，例如运行httpd容器可以使用命令。Kubernetes，通常简称为K

    阅读更多2024-11-15
  • Linux权限和开发工具(3)

    我们在做项目的时候可能会遇到对自己的修改不满意,想要回到上一个版本时候,就需要对自己完成一个阶段,对当前阶段进行备份,就方便我们后续进行版本回退了。后的程序可以执行但是文件也会大一些,让生成的程序带上

    阅读更多2024-11-15
  • C++ 编程基础（5）类与对象 | 5.8、面向对象五大原则

    在软件开发领域，面向对象编程（OOP）是一种重要的编程范式，它通过封装、继承和多态等特性，提高了代码的可重用性、灵活性和可维护性。C++作为一种强大的面向对象编程语言，充分体现了这些原则。在面向对象的

    阅读更多2024-11-15
  • Tailwind 安装使用

    Tailwind 安装使用

    阅读更多2024-11-15
  • JavaScript中的二叉树排序你了解吗？

    在计算机科学中，二叉树是一种常见的数据结构，用于存储和组织数据。二叉树排序（Binary Tree Sort）是一种基于二叉搜索树的排序算法。它的基本思想是将待排序的元素插入到二叉搜索树中，然后通过中

    阅读更多2024-11-15
自学内容网
Copyright © 2015-2024