碼上歡樂
相關內容    簡體    繁體

flink入門到實戰（2）flink優化總結

本文轉載自   查看原文   2020-05-07 11:01   2754    大數據之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


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 flink入門實戰總結 flink入門到實戰（3）flink進階篇 flink入門到實戰（5）flink流處理從0到1 flink入門到實戰（8）flink 有狀態 udf引發的大坑一 Flink原理、實戰與性能優化讀書筆記 Spark SQL入門到實戰之（8）數據傾斜優化 Flink（一）Flink的入門簡介 Canal詳細入門實戰(使用總結) Docker學習總結（1）——Docker實戰之入門以及Dockerfile（一） flink學習筆記-flink實戰
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM