flink入门到实战（2）flink优化总结

一、高性能Flink SQL优化技巧

1、Group Aggregate优化技巧

  • 开启MicroBatch或MiniBatch（提升吞吐）

    MicroBatch和MiniBatch都是微批处理，只是微批的触发机制略有不同。原理同样是缓存一定的数据后再触发处理，以减少对State的访问，从而提升吞吐并减少数据的输出量。

    MiniBatch主要依靠在每个Task上注册的Timer线程来触发微批，需要消耗一定的线程调度性能。MicroBatch是MiniBatch的升级版，主要基于事件消息来触发微批，事件消息会按您指定的时间间隔在源头插入。MicroBatch在元素序列化效率、反压表现、吞吐和延迟性能上都要优于MiniBatch。

    • 适用场景

      微批处理通过增加延迟换取高吞吐，如果您有超低延迟的要求，不建议开启微批处理。通常对于聚合的场景，微批处理可以显著的提升系统性能，建议开启。

      说明 MicroBatch模式也能解决两级聚合数据抖动问题。

    • 开启方式

      MicroBatch和MiniBatch默认关闭，开启方式如下。

      ​

      # 3.2及以上版本开启Window miniBatch方法（3.2及以上版本默认不开启Window miniBatch）。
sql.exec.mini-batch.window.enabled=true
# 批量输出的间隔时间，在使用microBatch策略时，需要增加该配置，且建议和blink.miniBatch.allowLatencyMs保持一致。
blink.microBatch.allowLatencyMs=5000
# 在使用microBatch时，需要保留以下两个miniBatch配置。
blink.miniBatch.allowLatencyMs=5000
# 防止OOM设置每个批次最多缓存数据的条数。
blink.miniBatch.size=20000

  • 开启LocalGlobal（解决常见数据热点问题）

    LocalGlobal优化将原先的Aggregate分成Local+Global两阶段聚合，即MapReduce模型中的Combine+Reduce处理模式。第一阶段在上游节点本地攒一批数据进行聚合（localAgg），并输出这次微批的增量值（Accumulator）。第二阶段再将收到的Accumulator合并（Merge），得到最终的结果（GlobalAgg）。

    LocalGlobal本质上能够靠LocalAgg的聚合筛除部分倾斜数据，从而降低GlobalAgg的热点，提升性能。您可以结合下图理解LocalGlobal如何解决数据倾斜的问题。img

    • 适用场景

      LocalGlobal适用于提升如SUM、COUNT、MAX、MIN和AVG等普通聚合的性能，以及解决这些场景下的数据热点问题。

      说明 开启LocalGlobal需要UDAF实现Merge方法。

    • 开启方式

      实时计算2.0版本开始，LocalGlobal是默认开启的，参数是blink.localAgg.enabled=true ，但是需要在microbatch或minibatch开启的前提下才能生效。

    • 判断是否生效

      观察最终生成的拓扑图的节点名字中是否包含GlobalGroupAggregate或LocalGroupAggregate。

  • 开启PartialFinal（解决COUNT DISTINCT热点问题）

    LocalGlobal优化针对普通聚合（例如SUM、COUNT、MAX、MIN和AVG）有较好的效果，对于COUNT DISTINCT收效不明显，因为COUNT DISTINCT在Local聚合时，对于DISTINCT KEY的去重率不高，导致在Global节点仍然存在热点。

    之前，为了解决COUNT DISTINCT的热点问题，通常需要手动改写为两层聚合（增加按Distinct Key取模的打散层）。自2.2.0版本开始，实时计算提供了COUNT DISTINCT自动打散，即PartialFinal优化，您无需自行改写为两层聚合。PartialFinal和LocalGlobal的原理对比参见下图。img

    • 适用场景

      使用COUNT DISTINCT，但无法满足聚合节点性能要求。

      说明

      • 不能在包含UDAF的Flink SQL中使用PartialFinal优化方法。
      • 数据量不大的情况下，不建议使用PartialFinal优化方法。PartialFinal优化会自动打散成两层聚合，引入额外的网络Shuffle，在数据量不大的情况下，浪费资源。

    • 开启方式

      默认不开启，使用参数显式开启blink.partialAgg.enabled=true。

    • 判断是否生效

      观察最终生成的拓扑图的节点名中是否包含Expand节点，或者原来一层的聚合变成了两层的聚合。

  • 改写为AGG WITH FILTER语法（提升大量COUNT DISTINCT场景性能）

    说明 仅实时计算2.2.2及以上版本支持AGG WITH FILTER语法。

    统计作业需要计算各种维度的UV，例如全网UV、来自手机客户端的UV、来自PC的UV等等。建议使用标准的AGG WITH FILTER语法来代替CASE WHEN实现多维度统计的功能。实时计算目前的SQL优化器能分析出Filter参数，从而同一个字段上计算不同条件下的COUNT DISTINCT能共享State，减少对State的读写操作。性能测试中，使用AGG WITH FILTER语法来代替CASE WHEN能够使性能提升1倍。

    • 适用场景

      建议您将AGG WITH CASE WHEN的语法都替换成AGG WITH FILTER的语法，尤其是对同一个字段上计算不同条件下的COUNT DISTINCT结果，性能提升很大。

    • 原始写法

      ​

      COUNT(distinct visitor_id) as UV1 , COUNT(distinct case when is_wireless='y' then visitor_id else null end) as UV2

    • 优化写法

      ​

      COUNT(distinct visitor_id) as UV1 , COUNT(distinct visitor_id) filter (where is_wireless='y') as UV2

2、TopN优化技巧

  • TopN算法

    当TopN的输入是非更新流（例如Source），TopN只有一种算法AppendRank。当TopN的输入是更新流时（例如经过了AGG/JOIN计算），TopN有3种算法，性能从高到低分别是：UpdateFastRank 、 UnaryUpdateRank和RetractRank。算法名字会显示在拓扑图的节点名字上。

    • UpdateFastRank ：最优算法。

      需要具备2个条件：

      • 输入流有PK信息。

      • 排序字段的更新是单调的，且单调方向与排序方向相反。例如，ORDER BY COUNT/COUNT_DISTINCT/SUM（正数）DESC（仅实时计算2.2.2及以上版本支持）。

        如果您要获取到优化Plan，需要告诉系统total_fee

        ​

        insert
  into print_test
SELECT
  cate_id,
  seller_id,
  stat_date,
  pay_ord_amt  --不输出rownum字段，能减小结果表的输出量。
FROM (
    SELECT
      *,
      ROW_NUMBER () OVER (
        PARTITION BY cate_id,
        stat_date  --注意要有时间字段，否则state过期会导致数据错乱。
        ORDER
          BY pay_ord_amt DESC
      ) as rownum  --根据上游sum结果排序。
    FROM (
        SELECT
          cate_id,
          seller_id,
          stat_date,
          --重点。声明Sum的参数都是正数，所以Sum的结果是单调递增的，因此TopN能使用优化算法，只获取前100个数据。
          sum (total_fee) filter (
            where
              total_fee >= 0
          ) as pay_ord_amt
        FROM
          random_test
        WHERE
          total_fee >= 0
        GROUP
          BY cate_name,
          seller_id,
          stat_date
      ) a
    WHERE
      rownum <= 100
  );

    • UnaryUpdateRank：仅次于UpdateFastRank的算法。需要具备1个条件：输入流中存在PK信息。例如，ORDER BY AVG。

    • RetractRank：普通算法，性能最差，不建议在生产环境使用该算法。请检查输入流是否存在PK信息，如果存在，则可进行UnaryUpdateRank或UpdateFastRank优化。

  • TopN优化方法

    • 无排名优化

      TopN的输出结果无需要显示rownum值，仅需在最终前端显式时进行1次排序，极大地减少输入结果表的数据量。无排名优化方法详情请参见TopN语句。

    • 增加TopN的Cache大小

      TopN为了提升性能有一个State Cache层，Cache层能提升对State的访问效率。TopN的Cache命中率的计算公式为。

      ​

      cache_hit = cache_size*parallelism/top_n/partition_key_num

      ​ 例如，Top100配置缓存10000条，并发50，当您的PatitionBy的key维度较大时，例如10万级别时，Cache命中率只有10000*50/100/100000=5%，命中率会很低，导致大量的请求都会击中State（磁盘），性能会大幅下降。因此当PartitionKey维度特别大时，可以适当加大TopN的CacheS ize，相对应的也建议适当加大TopN节点的Heap Memory（请参见

      手动配置调优

      ）。

      ​

      ##默认10000条，调整TopN cahce到20万，那么理论命中率能达200000*50/100/100000 = 100%。
blink.topn.cache.size=200000

    • PartitionBy的字段中要有时间类字段

      例如每天的排名，要带上Day字段。否则TopN的结果到最后会由于State ttl有错乱。

3、高效去重方案

说明 仅实时计算3.2.1及以上版本支持高效去重方案。

实时计算的源数据在部分场景中存在重复数据，去重成为了用户经常反馈的需求。实时计算有保留第一条（Deduplicate Keep FirstRow）和保留最后一条（Deduplicate Keep LastRow）2种去重方案。

  • 语法

    由于SQL上没有直接支持去重的语法，还要灵活的保留第一条或保留最后一条。因此我们使用了SQL的ROW_NUMBER OVER WINDOW功能来实现去重语法。去重本质上是一种特殊的TopN。

    ​

    SELECT *
FROM (
   SELECT *,
    ROW_NUMBER() OVER ([PARTITION BY col1[, col2..]
     ORDER BY timeAttributeCol [asc|desc]) AS rownum
   FROM table_name)
WHERE rownum = 1
    参数说明
    ROW_NUMBER() 计算行号的OVER窗口函数。行号从1开始计算。
    PARTITION BY col1[, col2..] 可选。指定分区的列，即去重的KEYS。
    ORDER BY timeAttributeCol [asc|desc]) 指定排序的列，必须是一个时间属性的字段（即Proctime或Rowtime）。可以指定顺序（Keep FirstRow）或者倒序 （Keep LastRow）。
    rownum 仅支持rownum=1或rownum<=1。

    ​ 如上语法所示，去重需要两层Query：

    1. 使用

      ROW_NUMBER()

      窗口函数来对数据根据时间属性列进行排序并标上排名。

      说明

      • 当排序字段是Proctime列时，Flink就会按照系统时间去重，其每次运行的结果是不确定的。
      • 当排序字段是Rowtime列时，Flink就会按照业务时间去重，其每次运行的结果是确定的。

    2. 对排名进行过滤，只取第一条，达到了去重的目的。

      说明 排序方向可以是按照时间列的顺序，也可以是倒序：

  • Deduplicate Keep FirstRow

    保留首行的去重策略：保留KEY下第一条出现的数据，之后出现该KEY下的数据会被丢弃掉。因为STATE中只存储了KEY数据，所以性能较优，示例如下。

    ​

    SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b ORDER BY proctime) as rowNum
  FROM T
)
WHERE rowNum = 1

    说明 以上示例是将T表按照b字段进行去重，并按照系统时间保留第一条数据。Proctime在这里是源表T中的一个具有Processing Time属性的字段。如果您按照系统时间去重，也可以将Proctime字段简化PROCTIME()函数调用，可以省略Proctime字段的声明。

  • Deduplicate Keep LastRow

    保留末行的去重策略：保留KEY下最后一条出现的数据。保留末行的去重策略性能略优于LAST_VALUE函数，示例如下。

    ​

    SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, d ORDER BY rowtime DESC) as rowNum
  FROM T
)
WHERE rowNum = 1

    说明 以上示例是将T表按照b和d字段进行去重，并按照业务时间保留最后一条数据。Rowtime在这里是源表T中的一个具有Event Time属性的字段。

4、高效的内置函数

  • 使用内置函数替换自定义函数

    实时计算的内置函数在持续的优化当中，请尽量使用内部函数替换自定义函数。实时计算2.0版本对内置函数主要进行了如下优化：

    • 优化数据序列化和反序列化的耗时。
    • 新增直接对字节单位进行操作的功能。

  • KEY VALUE函数使用单字符的分隔符

    KEY VALUE 的签名：KEYVALUE(content, keyValueSplit, keySplit, keyName)，当keyValueSplit和KeySplit是单字符（例如，冒号（:）、逗号（,））时，系统会使用优化算法，在二进制数据上直接寻找所需的keyName 的值，而不会将整个content做切分。性能约提升30%。

  • 多KEY VALUE场景使用MULTI_KEYVALUE

    说明 仅实时计算2.2.2及以上版本支持MULTI_KEYVALUE。

    ​ 在Query中对同一个Content进行大量KEY VALUE的操作，会对性能产生很大影响。例如Content中包含10个Key-Value对，如果您希望把10个Value的值都取出来作为字段，您就需要写10个KEY VALUE函数，则系统就会对Content进行10次解析，导致性能降低。

    在这种情况下，建议您使用MULTI_KEYVALUE表值函数，该函数可以对Content只进行一次Split解析，性能约能提升50%~100%。

  • LIKE操作注意事项

    • 如果需要进行StartWith操作，使用LIKE 'xxx%'。
    • 如果需要进行EndWith操作，使用LIKE '%xxx'。
    • 如果需要进行Contains操作，使用LIKE '%xxx%'。
    • 如果需要进行Equals操作，使用LIKE 'xxx'，等价于str = 'xxx'。
    • 如果需要匹配_字符，请注意要完成转义LIKE '%seller/id%' ESCAPE '/'。_在SQL中属于单字符通配符，能匹配任何字符。如果声明为LIKE '%seller_id%'，则不单会匹配seller_id还会匹配seller#id、sellerxid或seller1id 等，导致结果错误。

  • 慎用正则函数（REGEXP）

    正则表达式是非常耗时的操作，对比加减乘除通常有百倍的性能开销，而且正则表达式在某些极端情况下可能会进入无限循环

5、网络传输的优化

目前常见的Partitioner策略包括：

  • KeyGroup/Hash：根据指定的Key分配。

  • Rebalance：轮询分配给各个Channel。

  • Dynamic-Rebalance：根据下游负载情况动态选择分配给负载较低的Channel。

  • Forward：未Chain一起时，同Rebalance。Chain一起时是一对一分配。

  • Rescale：上游与下游一对多或多对一。

  • 使用Dynamic-Rebalance替代Rebalance

    Dynamic-Rebalance可以根据当前各Subpartition中堆积的Buffer的数量，选择负载较轻的Subpartition进行写入，从而实现动态的负载均衡。相比于静态的Rebalance策略，在下游各任务计算能力不均衡时，可以使各任务相对负载更加均衡，从而提高整个作业的性能。例如，在使用Rebalance时，发现下游各个并发负载不均衡时，可以考虑使用Dynamic-Rebalance。参数：task.dynamic.rebalance.enabled=true， 默认关闭。

  • 使用Rescale替代Rebalance

    说明 仅实时计算2.2.2及以上版本支持Rescale。

    例如，上游是5个并发，下游是10个并发。当使用Rebalance时，上游每个并发会轮询发给下游10个并发。当使用Rescale时，上游每个并发只需轮询发给下游2个并发。因为Channel个数变少了，Subpartition的Buffer填充速度能变快，能提高网络效率。当上游的数据比较均匀时，且上下游的并发数成比例时，可以使用Rescale替换Rebalance。参数：enable.rescale.shuffling=true，默认关闭。

6、推荐的优化配置方案

综上所述，作业建议使用如下的推荐配置。

​

# EXACTLY_ONCE语义。
blink.checkpoint.mode=EXACTLY_ONCE
# checkpoint间隔时间，单位毫秒。
blink.checkpoint.interval.ms=180000
blink.checkpoint.timeout.ms=600000
# 2.x使用niagara作为statebackend，以及设定state数据生命周期，单位毫秒。
state.backend.type=niagara
state.backend.niagara.ttl.ms=129600000
# 2.x开启5秒的microbatch。
blink.microBatch.allowLatencyMs=5000
# 整个Job允许的延迟。
blink.miniBatch.allowLatencyMs=5000
# 单个batch的size。
blink.miniBatch.size=20000
# local 优化，2.x默认已经开启，1.6.4需手动开启。
blink.localAgg.enabled=true
# 2.x开启PartialFina优化，解决COUNT DISTINCT热点。
blink.partialAgg.enabled=true
# union all优化。
blink.forbid.unionall.as.breakpoint.in.subsection.optimization=true
# object reuse优化，默认已开启。
#blink.object.reuse=true
# GC优化（SLS做源表不能设置该参数）。
blink.job.option=-yD heartbeat.timeout=180000 -yD env.java.opts='-verbose:gc -XX:NewRatio=3 -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:ParallelGCThreads=4'
# 时区设置。
blink.job.timeZone=Asia/Shanghai

二、Flink性能调优

1、自动配置调优

实时计算 Flink新增自动调优功能autoconf。能够在流作业以及上下游性能达到稳定的前提下，根据您作业的历史运行状况，重新分配各算子资源和并发数，达到优化作业的目的。

1、首次智能调优

  1. 创建一个作业。

  2. 上线作业。选择智能推荐配置，指定使用CU数为系统默认，不填即可。点击下一步。 img

  3. 数据检查，预估消耗CU数。 img

  4. 在运维界面启动作业，根据实际业务需要指定读取数据时间。 img

    说明:实时计算作业启动时候需要您指定启动时间。实际上就是从源头数据存储的指定时间点开始读取数据。指定读取数据时间需要在作业启动之前。例如，设置启动时间为1小时之前。

  5. 待作业稳定运行10分钟后，且以下状态符合要求，即可开始下一次性能调优。

    • 运行信息拓扑图中IN_Q不为100%。 img
    • 数据输入RPS符合预期。 img

2、非首次性能调优

  1. 停止>下线作业。 img
  2. 重新上线作业。选择智能推荐配置，指定使用CU数为系统默认，不填即可。点击下一步。 img
  3. 数据检查，再次预估消耗CU数。 img
  4. 在运维界面启动作业，待作业稳定运行十分钟后，即可再一次性能调优。 img

说明：

  • 自动配置调优一般需要3到5次才能达到理想的调优效果。请完成首次性能调优后，重复非首次性能调优过程多次。
  • 每次调优前，请确保足够的作业运行时长，建议10分钟以上。
  • 指定CU数（参考值） = 实际消耗CU数*目标RPS/当前RPS。
    • 实际消耗CU数：上一次作业运行时实际消耗CU
    • 目标RPS：输入流数据的实际RPS（或QPS）
    • 当前RPS：上一次作业运行时实际的输入RPS

2、手动配置调优

手动配置调优可以分以下三个类型。

  • 资源调优
  • 作业参数调优
  • 上下游参数调优

1、资源调优

资源调优即是对作业中的Operator的并发数（parallelism）、CPU（core）、堆内存（heap_memory）等参数进行调优。

1、分析定位资源调优节点

定位性能瓶颈节点

性能瓶颈节点为Vertex拓扑图最下游中参数IN_Q值为100%的一个或者多个节点。如下图，7号节点为性能瓶颈节点。 img

分析性能瓶颈因素

性能瓶颈的可分为三类。

  • 并发（parallelism）不足
  • CPU（core）不足
  • MEM（heap_memory）不足

如下图，7号节点的性能瓶颈是资源（CPU和/或MEM）配置不足所导致。 img

说明：判断性能瓶颈因素方法

  • 瓶颈节点的资源健康分为100，则认为资源已经合理分配，性能瓶颈是并发数不足所导致。
  • 瓶颈节点的资源健康分低于100，则认为性能瓶颈是单个并发的资源（CPU和/或MEM）配置不足所导致。
  • 无持续反压，但资源健康分低于100，仅表明单个并发的资源使用率较高，但暂不影响作业性能，可暂不做调优。

通过作业运维页面中Metrics Graph功能，进一步判断性能瓶颈是CPU不足还是MEM不足。步骤如下。

  1. 运维界面中，点击TaskExecutor，找到性能瓶颈节点ID，点击查看详情。 img
  2. 选择Metrics Graph，根据曲线图判断CPU或者MEM是否配置不足（很多情况下两者同时不足）。 img

2、调整资源配置

完成了性能瓶颈因素判断后，点击开发>基本属性>跳转到新窗口配置，开始调整资源配置。 img

批量修改Operator

  1. 点击GROUP框，进入批量修改Operator数据窗口。 img

    说明：

    1. GROUP内所有的operator具有相同的并发数。
    2. GROUP的core为所有operator的最大值。
    3. GROUP的_memory为所有operator之和。
    4. 建议单个Job维度的CPU:MEM=1:4，即1个核对应4G内存。

  2. 配置修改完成后点击应用当前配置并关闭窗口。 img

单个修改Operator

  1. 点击Operator框，进入修改Operator数据窗口。 img
  2. 配置修改完成后点击应用当前配置并关闭窗口。 img

参数调整说明

您只需调整parallelism、core和heap_memory三个参数，即能满足大部分的资源调优需求。

  • Parallelism
    • source节点 资源根据上游Partition数来。例如source的个数是16，那么source的并发可以配置为16、8、4等。不能超过16。
    • 中间节点 根据预估的QPS计算。对于数据量较小的任务，设置和source相同的并发度。QPS高的任务，可以配置更大的并发数，例如64、128、或者256。
    • sink节点 并发度和下游存储的Partition数相关，一般是下游Partition个数的2~3倍。如果配置太大会导致数据写入超时或失败。例如，下游sink的个数是16，那么sink的并发最大可以配置48。
  • Core 即CPU，根据实际CPU使用比例配置，建议配置值为0.25，可大于1。
  • Heap_memory 堆内存。根据实际内存使用状况进行配置。
  • 其他参数
    • state_size：默认为0，group by、join、over、window等operator需设置为1。
    • direct_memory：JVM堆外内存，默认值为0, 建议不要修改。
    • native_memory：JVM堆外内存，默认值为0，建议修改为10MB。
    • chainingStrategy：chain策略，根据实际需要修改。

2、作业参数调优

  1. 在开发页面的右侧选择作业参数。 img
  2. 输入调优语句。
优化解决问题调优语句
MiniBatch 提升吞吐，降低对下游压力仅对Group by有效。 blink.miniBatch.allowLatencyMs=5000 blink.miniBatch.size=1000
LocalGlobal 优化数据倾斜问题 blink.localAgg.enable=true
TTL 设置State状态时间 1.x：state.backend.rocksdb.ttl.ms=129600000 2.x：state.backend.niagara.ttl.ms=129600000 其中，1.x 表示需显式开启，2.x 表示默认开启。

注意：添加或删除MiniBatch或LocalGlobal参数，job状态会丢失，修改值大小状态不会丢失。

3、上下游参数调优

实时计算 Flink可以在with参数内设置相应的参数，达到调优上下游存储性能的目的。

调优步骤：

  1. 进入作业的开发界面。
  2. 确定需要调优的上下游引用表的语句。
  3. 在with参数中配置相应的调优参数。如下图。 img

1、batchsize参数调优

实时计算 Flink的每条数据均会触发上下游存储的读写，会对上下游存储形成性能压力。可以通过设置batchsize，批量的读写上下游存储数据来降低对上下游存储的压力。

名字参数详情设置参数值
Datahub源表 batchReadSize 单次读取条数 可选，默认为10
Datahub结果表 batchSize 单次写入条数 可选，默认为300
日志服务源表 batchGetSize 单次读取logGroup条数 可选，默认为10
ADB结果表 batchSize 每次写入的批次大小 可选，默认为1000
RDS结果表 batchSize 每次写入的批次大小 可选，默认为50

注意： 添加、修改或者删除以上参数后，作业必须停止-启动后，调优才能生效。

2、cache参数调优

名字参数详情设置参数值
RDS维表 Cache 缓存策略 默认值为None,可选LRU、ALL。
RDS维表 cacheSize 缓存大小 默认值为None,可选LRU、ALL。
RDS维表 cacheTTLMs 缓存超时时间 默认值为None,可选LRU、ALL。
OTS维表 Cache 缓存策略 默认值为None, 可选LRU,不支持ALL。
OTS维表 cacheSize 缓存大小 默认值为None, 可选LRU,不支持ALL。
OTS维表 cacheTTLMs 缓存超时时间 默认值为None, 可选LRU,不支持ALL。

注意： 添加、修改或者删除以上参数后，作业必须停止-启动后，调优才能生效。

4、手动配置调优流程

  1. 资源调优、作业参数调优、上下游参数调优
  2. 开发上线作业
  3. 资源配置方式：使用上次资源配置
  4. 数据检查
  5. 上线

说明：完成资源、作业参数、上下游参数调优后，手动配置调优后续的步骤与自动配置调优基本一致。区别在于资源配置环节需要选择使用上次资源配置。 img

3、FAQ

1、Q：性能调优后作业为什么运行不起来？

A:可能性1：首次自动配置时指定了CU数，但指定的CU数太小（比如小于自动配置默认算法的建议值，多见于作业比较复杂的情况），建议首次自动配置时不指定CU数。 可能性2：默认算法建议的CU数或指定的CU数超过了项目当前可用的CU数，建议扩容。

2、Q：Vertex拓扑中看不到持续反压，但延迟却非常大，为什么？

A：可能性1：若延时直线上升，需考虑是否上游source中部分partition中没有新的数据，因为目前delay统计的是所有partition的延时最大值。 可能性2：Vertex拓扑中看不到持续反压，那么性能瓶颈点可能出现在source节点。因为source节点没有输入缓存队列，即使有性能问题，IN_Q也永远为0（同样，sink节点的OUT_Q也永远为0）。 解决方案：通过手动配置调优，将source节点（GROUP）中的operator中chainingStrategy修改为HEAD，将其从GROUP中拆解出来。然后上线运行后再看具体是哪个节点是性能瓶颈节点，若依然看不到性能瓶颈节点，则可能是因为上游source吞吐不够，需考虑增加source的batchsize或增加上游source的shard数。 imgimg

3、Q:如何判断持续反压，反压时如何判断性能瓶颈点？

A:Vertex拓扑中某些节点的IN_Q持续为100%则存在持续反压情况，最后一个(或多个)IN_Q为100%的节点为性能瓶颈点。如下示例： img 上图存在反压,瓶颈在6号节点。

img 上图存在反压，瓶颈在2号节点。

img 上图存在反压，瓶颈在8号节点。

img 上图可能存在节点，瓶颈在0号节点。

4、Q: 如何判断数据倾斜？

A：(1)表象上看，某些节点不论增加多大的并发仍存在性能瓶颈，则可能存在数据倾斜。 (2)在Vertex拓扑中点击疑似存在数据倾斜的节点（一般为性能瓶颈节点），进入SubTask List界面，重点观察RecvCnt和InQueue，若各ID对应的RecvCnt值差异较大（一般超过1个数量级）则认为存在数据倾斜，若个别ID的InQueue长期为100%，则认为可能存在数据倾斜。 解决方案：请您参看GROUP BY 数据出现热点、数据倾斜。 img

5、Q: 上线时指定15CU，但是上线后实际消耗仅为10CU，什么原因？

A：这种问题一般发生在Vertex只有一个节点的情况，此时由于source上游的物理表的shard数为1，Flink要求source的并发不能超过上游shard数，导致无法增加并发，因此亦无法增加到指定的CU数。 解决方案：

  1. 增加上游物理表的shard数。
  2. 将ID0的节点中的operator拆开，将source节点（GROUP）中的operator chainingStrategy修改为HEAD，将其从GROUP中拆解出来，然后手动配置调优。

6、Q: 上线时出现如左上图的告警，或出现诸如“Cannot set chaining strategy on Union Transformation”错误，如何处理？

A：这是由于作业的SQL有改动，导致DAG改变。 解决方案：通过重新获取配置解决，开发-基本属性-跳转到新窗口配置-重新获取配置信息。 img

img

作者：大码王

-------------------------------------------

个性签名：独学而无友，则孤陋而寡闻。做一个灵魂有趣的人！

如果觉得这篇文章对你有小小的帮助的话，记得在右下角点个“推荐”哦，博主在此感谢！

万水千山总是情，打赏一分行不行，所以如果你心情还比较高兴，也是可以扫码打赏博主，哈哈哈(っ•?ω•?)っ???！

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/huanghanyu/p/12841561.html