# Flink作业性能调优
#### 性能调优相关基本概念
- 消费组积压 消费组积压可通过topic最新数据offset减去该消费组已提交最大offset计算得出，说明的是该消费组当前待消费的数据总量。
  如果Flink作业对接的是kafka专享版，则可通过云监控服务(CES)进行查看。具体可选择"云服务监控 \> 分布式消息服务 \> kafka专享版" ，单击"kafka实例名称 \> 消费组" ，选择具体的消费组名称，查看消费组的指标信息。
  图1消费组   
  ![](https://support.huaweicloud.com/dli_faq/zh-cn_image_0000001082041926.png "点击放大") 
- 反压状态 反压状态是通过周期性对taskManager线程的栈信息采样，计算被阻塞在请求输出Buffer的线程比率来确定，默认情况下，比率在0.1以下为OK，0.1到0.5为LOW，超过0.5则为HIGH。
  
- 时延 Source端会周期性地发送带当前时间戳的LatencyMarker，下游算子接收到该标记后，通过当前时间减去标记中带的时间戳的方式，计算时延指标。算子的反压状态和时延可以通过Flink UI或者作业任务列表查看，一般情况下反压和高时延成对出现：
  图2反压状态和时延   
  ![](https://support.huaweicloud.com/dli_faq/zh-cn_image_0000001129079507.png "点击放大") 
 
#### 性能分析
由于Flink的反压机制，流作业在存在性能问题的情况下，会导致数据源消费速率跟不上生产速率，从而引起Kafka消费组的积压。在这种情况下，可以通过算子的反压和时延，确定算子的性能瓶颈点。
- 作业最后一个算子(Sink)反压正常（绿色），前面算子反压高（红色） ![](https://support.huaweicloud.com/dli_faq/zh-cn_image_0000001161715923.png)
  该场景说明性能瓶颈点在sink，此时需要根据具体数据源具体优化，比如对于JDBC数据源，可以通过调整写出批次(connector.write.flush.max-rows)、JDBC参数重写(rewriteBatchedStatements=true)等进行优化。
  
- 作业非倒数第二个算子反压高（红色） ![](https://support.huaweicloud.com/dli_faq/zh-cn_image_0000001115396324.png)
  该场景说明性能瓶颈点在Vertex2算子，可以通过查看该算子描述，确认该算子具体功能，以进行下一步优化。
  
- 所有算子反压都正常（绿色），但存在数据堆积 ![](https://support.huaweicloud.com/dli_faq/zh-cn_image_0000001115399366.png)
  该场景说明性能瓶颈点在Source，主要是受数据读取速度影响，此时可以通过增加Kafka分区数并增加source并发解决。
  
- 作业一个算子反压高（红色），而其后续的多个并行算子都不存在反压（绿色） ![](https://support.huaweicloud.com/dli_faq/zh-cn_image_0000001161799601.png)
  该场景说明性能瓶颈在Vertex2或者Vertex3，为了进一步确定具体瓶颈点算子，可以在FlinkUI页面开启inPoolUsage监控。如果某个算子并发对应的inPoolUsage长时间为100%，则该算子大概率为性能瓶颈点，需分析该算子以进行下一步优化。
  图3inPoolUsage监控   
  ![](https://support.huaweicloud.com/dli_faq/zh-cn_image_0000001161719815.png "点击放大") 
 
#### 性能调优
- rocksdb状态调优 topN排序、窗口聚合计算以及流流join等都涉及大量的状态操作，因而如果发现这类算子存在性能瓶颈，可以尝试优化状态操作的性能。主要可以尝试通过如下方式优化：
  - 增加状态操作内存，降低磁盘IO
    - 增加单slot cu资源数
    
    - 配置优化参数：
      - taskmanager.memory.managed.fraction=xx
      
      - state.backend.rocksdb.block.cache-size=xx
      
      - state.backend.rocksdb.writebuffer.size=xx
       
     
  
  - 开启微批模式，避免状态频繁操作 配置参数：
    - table.exec.mini-batch.enabled=true
    
    - table.exec.mini-batch.allow-latency=xx
    
    - table.exec.mini-batch.size=xx
     
  
  - 使用超高IO本地盘规格机型，加速磁盘操作
   
- group agg单点及数据倾斜调优 按天聚合计算或者group by key不均衡场景下，group聚合计算存在单点或者数据倾斜问题，此时，可以通过将聚合计算拆分成Local-Global进行优化。配置方式为设置调优参数: table.optimizer.aggphase-strategy=TWO_PHASE
  
- count distinct优化
  - 在count distinct关联key比较稀疏场景下，即使使用Local-Global，单点问题依然非常严重，此时可以通过配置以下调优参数进行分桶拆分优化:
    - table.optimizer.distinct-agg.split.enabled=true
    
    - table.optimizer.distinct-agg.split.bucket-num=xx
     
  
  - 使用filter替换case when: 例如：
    ```
    COUNT(DISTINCT CASE WHEN flag IN ('android', 'iphone')THEN user_id ELSE NULL END) AS app_uv
    ```
    可调整为
    ```
    COUNT(DISTINCT user_id) FILTER(WHERE flag IN ('android', 'iphone')) AS app_uv
    ```
    
   
- 维表join优化 维表join根据左表进入的每条记录join关联键，先在缓存中匹配，如果匹配不到，则从远程拉取。因而，可以通过如下方式优化:
  - 增加JVM内存并增加缓存记录条数
  
  - 维表设置索引，加快查询速度