Consumer assignor 查核點討論

[harryteng9527]

叢集規模

• 6 台節點作Brokers，9 台節點作 Client 端
• 叢集內 partitions 數量約 100~200 個

比較對象

• Kafka default assignor - Range assignor

查核點

• Consumer group 整體吞吐量上升 15 %
• 平均讀取的端對端延遲較 default assignor 下降 15 %
• 使用 CostAware assignor 的 consumers 需消費的總資料量較 Range assignor 的 consumers 平衡 15%

備註

1. 整體吞吐量上升是指Consumer group 的吞吐量
2. 第三個查核點的計算方式：分別使用預設、CostAware assignor 後，比較 (被分配最多資料的 consumer - 被分配最少資料量的 consumer) 的差距

[chia7712]

Consumer group 整體吞吐量上升 10 %

你有打算做什麼優化讓吞吐量有所差異嗎？我記得當前目標是以資料量的平衡為主，那在這個條件下應該很難看到吞吐量的提升

Lag 的資料量較 default assignor 少 5 %

請問這邊的 lag 是怎麼計算？

[harryteng9527]

你有打算做什麼優化讓吞吐量有所差異嗎？我記得當前目標是以資料量的平衡為主，那在這個條件下應該很難看到吞吐量的提升

我讓吞吐量提升的想法如下，再麻煩學長看一下有沒有合理 or 正常使用的情況下會不會發生這種問題：

目前負載定義為：流入 partition 的資料量，之後會為 NetworkCost 添加 PartitionCost

負載不均時，Consumer 的消費能力可能會是一個上限，例如下圖的例子，假設 consumer 消費資料的上限是 80 MB/s，但是 Consumer A 被分配了兩個 60 MB/s 的 partitions，導致 Consumer A 即使全力在讀取，吞吐量還是沒辦法達到 120 MB/s，然後 Consumer B 比較悠閒，只被分配到 20 MB/s 的工作量

透過流入 partition 的資料量來分配 partition 的話，就可以產生例如下圖的分配，這樣每個 consumer 都有 70 MB/s 的資料量要讀取，也不會因為消費能力的上限而消化不了 partition 的資料

這邊的 lag 是怎麼計算？

想透過 JMX 去撈 Consumer 的 Lag records - records-lag-max

[chia7712]

目前負載定義為：流入 partition 的資料量，之後會為 NetworkCost 添加 PartitionCost

這個方式可以

想透過 JMX 去撈 Consumer 的 Lag records - records-lag-max

如果已經有Admin的話，可以透過consumerGroups來拿取group offset，就不需要用 metrics

另外 lag 的部分跟吞吐量太相關，只是換個角度看，可能需要別種查核項目，例如各個 consumer 處理的資料量平衡

縮短 Downtime 時間 5 %

這個技術上的門檻比較多，可以參考我上面說的另一個查核點

[harryteng9527]

可以透過consumerGroups來拿取group offset，就不需要用 metrics

我當初是想說能以 Grafana 來呈現查核點就以 Grafana 來呈現，因為如果是用 Admin 的話是不是就要用 fileWriter 寫成 csv 檔再繪製圖表，要再加工一次。
雖然之後想看 partition 差異也要用 csv 來繪製圖表

各個 consumer 處理的資料量平衡

想詢問這個查核項目會不會也跟吞吐量很相關？

因為以我上面優化吞吐量的例子，我是靠分配某些負載較重的 partition 給 loading 較輕的 consumer，所以照理來說這樣分配完後，吞吐量會上升的原因就是 consumer 處理的資料量比較平衡

[chia7712]

想詢問這個查核項目會不會也跟吞吐量很相關？

這個查核點可以看的是「總體」資料的處理時間差異，情境會跟吞吐量不太一樣。例如在頻寬都足夠的狀態下，資料量分配不均就會導致「整體」資料的處理時間有所差異。

[harryteng9527]

學長好，
我想詢問週一討論時有提到的其中一個查核點，可以將改善整體吞吐量改成 consumer 個別提昇吞吐量 xx % ，但是我後來想了想後，感覺不能個別提昇，原因如下

以我上面提到我讓吞吐量上升的方法 #1437 (comment) ，以下都是以理想的情況討論

因為使用 Default assignor 時的分配會讓 consumer 無法負荷所有的流量(假設 consumer 只能消費 80MB/s 的資料)，這會讓 consumer group 整體的吞吐量為 100 MB/s。若是使用 NetworkIngress assignor，能夠依照流入 partition 的流量來分配，所以 consumer group 的整體吞吐量為 140 MB/s。如下圖，從 consumer group 來看，整體的吞吐量是上升的

但如果單看 group 內個別的 consumer，會發現 consumer A 其實是下降的，因為原本處理 80 MB/s 的資料，經由平衡後，吞吐量只會到 70 MB/s，下降了 10 MB/s。

而 consumer B 是上升的，因為將某個 loading 比較重的 partition 分配給 consumer B 。如下圖，Consumer A 吞吐量下降，而 Consumer B 吞吐量上升

[chia7712]

我想詢問週一討論時有提到的其中一個查核點，可以將改善整體吞吐量改成 consumer 個別提昇吞吐量 xx %

那就做整體的提升，這也是一個合理的情境，就是改成以"應用“(consumer group)的角度來看效能提升

[harryteng9527]

還想詢問學長 各個 Consumer 處理的資料量平衡 可否用 performance tool 結束的時間來呈現

因為 Performance tool 的 consumer 消費完 producer 送到 partition 的資料才會關閉(透過 read idle 來決定要不要關閉 consumer)，所以可以觀察執行時間來判斷有沒有平衡，因為夠平衡的話，performance tool 會比較快結束

[chia7712]

還想詢問學長 各個 Consumer 處理的資料量平衡 可否用 performance tool 結束的時間來呈現

由於最後要拉圖表的關係，我建議看看 metrics 中有沒有針對特定group 的 offset 紀錄，然後把 offset 拉成曲線圖，這樣不只可以看執行時間，也可以看過程的增加服務變化

[chia7712]

各個 Consumer 處理的資料量平衡

這個有改善程度可以寫嗎？

各 consumers 處理的資料量可看作每個 consumer 被分配到的負載相同，故執行結束時間相近

這句話是補充資料量為何要平衡嗎？

[harryteng9527]

我想再與學長確認一下，之前所說的每個查核點可以用不同的環境測試是指我測試不同查核點的時候，營造的 partition throttle 流量、consumer 參數都可以不同嗎？

例如我測試吞吐量提升的查核點時設計的 throttle 流量可以和測試延遲查核點、資料量平衡的查核點所設計的 throttle 流量可以不同

舉例來說就是測試 A 查核點時，流入 partition-0, partition-1 的流量是 30MB/s、100MB/s，調整 consumer 某個參數成 xx 值。而測試 B 查核點時，流入 partition-0, partition-1 的流量是 100MB/s、100MB/s ，調整 consumer 某某參數成 yy 值這樣

[chia7712]

之前所說的每個查核點可以用不同的環境測試是指我測試不同查核點的時候，營造的 partition throttle 流量、consumer 參數都可以不同嗎？

可以，只是不同情境的話要先明確寫上去

[harryteng9527]

這句話是補充資料量為何要平衡嗎？

對，應該是處理的資料量相同可看作每個 consumer 被分配到的負載相同

這個有改善程度可以寫嗎？

應該是可以，若要看的話應該是看 consumers 中，消費最多的資料量的 consumer 減去 消費最少的資料量的 consumer

[chia7712]

若要看的話應該是看 consumers 中，消費最多的資料量的 consumer 減去 消費最少的資料量的 consumer

ok 麻煩補到描述裡面

[chia7712]

第三個查核點的量化有想法了嗎

[harryteng9527]

第三個查核點的量化有想法了嗎

現在有用 consumer JMX 暴露出來的 Lag metric 來觀測，應該會呈現此 metric 來量化優化程度

[harryteng9527]

現在有使用現在正在實作的 Cost aware assignor #1524 來做實驗

目前使用 10 個 topic，隨機產生 partition 數量，partition 的數量為 1~16

今天有做一次實驗，查核點的數據如下：

• 吞吐量提昇 12.5%
• 延遲降低 5%
• Lag 降低 35%

實驗環境

• 6 台 broker
• 10 個 topics，總共 76 個 partition
• 9 個 producer，Key distribution 使用 Zipfian 分佈，營造某些 partition 較熱門的情境
• 6 個 consumer，分別使用 Range assignor 與 CostAware assignor
  • CostAware 的 upper bound 為 40 MiB, traffic interval 為 10MiB

實驗數據

這張圖是運行十分鐘的實驗，每條線代表使用不同 assignor 的 consumer group 整體(6 個 consumers)吞吐量，各 assignor 的平均吞吐量如下：

• Range assignor : 1.172 GiB
• CostAware assignor : 1.319 GiB

---

想順便問學長一些問題：

1. 查核時，應該要使用多少 partitions 的情境
2. 我有點想改變延遲的計算公式，目前的計算方式為 [ASSIGNOR] Implement CostAwareAssignor #1524 (comment) ，但是想不到有什麼好的計算方法，想詢問學長有沒有什麼常用來計算 latency 的公式
3. 目前計算改善 Lag 多少 percent的方式是將所有 partitions 的 lag 加總然後相除，不知道是否合理，還是有更好的算法

[chia7712]

應該要使用多少 partitions 的情境

你這個題目適合的數量是 topics 雙位數，加總後的partitions 三位數，例如總共10個topics 每個各自10～20 partitions

[chia7712]

想詢問學長有沒有什麼常用來計算 latency 的公式

E2E是一個比較適合的指標，perf tool 應該已經有相關的指標

[chia7712]

目前計算改善 Lag 多少 percent的方式是將所有 partitions 的 lag 加總然後相除，不知道是否合理，還是有更好的算法

延遲的部分（第二個查核項目）可以有幾個「角度」，第一個當然是E2E、第二個則是落後多少「筆」資料，但前者更能反應 streaming 的延遲

所以我比較建議是看能否抓出不同consumer的e2e 後，看平均延遲就好

[harryteng9527]

已更新查核項目，再麻煩 review，謝謝
目前取得延遲數據的部份 performance tool 還沒有支援，之後會完成