Apache Pulsar 微信大流量實時推薦場景下實踐詳解

2022-11-13 14:01:26

導語

本文整理自 8 月 Apache Pulsar Meetup 上，劉燊題為《Apache Pulsar 在微信的大流量實時推薦場景實踐》的分享。本文介紹了微信團隊在大流量場景下將 Pulsar 部署在 K8s 上的實踐與優化、非持久化 Topic 的應用、負載均衡與 Broker 快取優化實踐與COS Offloader 開發與應用。

作者簡介

劉燊，騰訊微信高階研發工程師，Apache Pulsar Contributor。

在通訊社交領域，微信已經成為國內當之無愧的社交霸主。使用者人數在 2018 年突破了 10 億，截至 2021 年第三季度末，微信每月活動賬戶總數已達到 12.6 億人，可以說，微信已經成為國人生活的一部分。

微信的業務場景包括推薦業務、風控、監控系統、AI 平臺等。資料通過 SDK 和資料採集方式接入，經由 MQ、Kafka、Pulsar 訊息中介軟體，其中 Pulsar 發揮了很大的作用。中介軟體下游接入資料計算層 Hadoop、Spark、Flink、ClickHouse、TensorFlow 等計算平臺，由於本次介紹實時推薦場景，因此較多使用 Flink 和 TensorFlow。落地儲存平臺則包括 HDFS、HBase、Redis 以及各類自研 KV。

團隊選型 Pulsar 的初期目標是獲得一個滿足巨量資料流量場景並且運維管理便捷的訊息佇列系統。最終選擇 Pulsar 的主要原因有五點：

在騰訊自研上雲的大背景下，團隊非常看重雲原生特性。Pulsar 的雲原生特性，包括分散式、彈性伸縮、讀寫分離等都體現出優勢。Pulsar 邏輯層 Broker 無狀態，直接提供服務。儲存層 Bookie 有狀態，但是節點對等，且 Bookie 自帶多副本容災；
Pulsar 支援資源隔離，可以軟隔離或硬隔離，避免不同業務之間互相影響；
Pulsar 支援靈活的 Namespace/Topic 策略管控，對叢集的管理和維護有很大幫助；
Pulsar 能夠便捷擴容，邏輯層 Broker 的無狀態和負載均衡策略允許快速擴容，儲存層 Bookie 節點之間互相對等也便於快速擴容，可以輕鬆應對流量暴漲場景；
Pulsar 具備多語言使用者端能力，微信的業務場景中涉及 C/C++、TensorFlow、Python 等語言，Pulsar 可以滿足需求。

實踐 1：大流量場景下的 K8s 部署實踐

微信團隊使用了 Pulsar 官網提供的 K8s Helm chart 部署方式。

原生部署架構中，流量從 Proxy 代理層進入，經過 Broker 邏輯服務層寫入 Bookie 儲存層。Proxy 代理層代理使用者端和 Broker 之間的連線，Broker 層管理 Topic，Bookie 層負責持久化訊息儲存。在上圖中，入流量和出流量分別用 In 和 Out 進行標記，Replica 是設定的副本。

在應用的過程中團隊發現了兩個問題：首先 Proxy 代理了 Pulsar 使用者端的請求，導致 Broker 無法獲取使用者端 IP，增加了運維難度；其次，當叢集流量較大時，叢集內部頻寬會成為瓶頸。上圖架構內，叢集入流量為（2+ 副本數）倍；出流量最大為 3 倍，Consumer、Proxy、Broker 和 Bookie 間分別有一倍流量，但是僅極端情況下流量會全量從 Bookie 流出。假設出入流量都是 10 GBps，副本數為 3，叢集內入流量會放大為 50 GBps，出流量會放大為 30 GBps。另外預設情況下 Proxy 服務只有一個負載均衡器承載所有流量，壓力巨大。

這裡可以看出瓶頸主要出現在 Proxy 層，該層造成了很大流量浪費。而 Pulsar 實際上支援 Broker 直連，因此團隊在此基礎上進行了一些優化：

團隊利用了騰訊雲 K8s 叢集的能力，給 Broker 設定了彈性網路卡，並使 Broker 的 IP 直接暴露在叢集外，可以被外部使用者端直接存取。Broker 服務也設定了負載均衡器。這樣使用者端可以直接存取負載均衡器 IP，再經過 Pulsar 內部協定的 Lookup 操作找到要存取的 Topic 所處的 Broker。由此節省了 Proxy 帶來的額外頻寬消耗。

團隊在 K8s 部署方面還做了以下優化工作：

如上文所述去 Proxy；
Bookie 使用多盤多目錄 + 本地 SSD 提升效能，由於原社群版本 Pulsar 不支援多盤多目錄，這裡團隊做了改進支援併合併入社群(github.com/apache/puls…)；
紀錄檔採集使用騰訊雲 CLS（紀錄檔服務），統一的紀錄檔服務可以簡化分散式多節點系統的運維、問題查詢操作；
指標採集使用 Grafana + Kvass + Thanos，預設指標採集使用了單機服務，很快出現了效能瓶頸，優化後問題解決且支援水平擴容。

實踐 2：非持久化 Topic 的應用

生產者和消費者是同 Broker 中的 Dispatcher 模組互動的，而持久化 Topic 中生產者資料會通過 Dispatcher 進入 Managed Ledger 模組，再呼叫 Bookie 使用者端與 Bookie 互動。非持久化 Topic 中資料不會進入 Managed Ledger，而是直接傳送給消費者。在大流量場景中，非持久化 Topic 由於不需要與 Bookie 互動，對叢集的頻寬壓力會明顯降低。

非持久化 Topic 在大流量實時推薦場景中有應用，但具體的應用場景必須滿足“可容忍少量資料丟失”的要求。實踐中有三種場景滿足這一要求：

大流量 + 消費端處理能力不足的實時訓練任務；
時效性敏感的實時訓練任務；
抽樣評測任務。

實踐 3：負載均衡與 Broker 快取優化

以上是一個線上真實的場景。生產環境中出現了反覆 bundle unload 的問題，導致 Broker 負載反覆波動。

該場景中使用了以下負載均衡設定：

loadManagerClassName=org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl
loadBalancerLoadSheddingStrategy=org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder
loadBalancerBrokerThresholdShedderPercentage=10
loadBalancerBrokerOverloadedThresholdPercentage=70
Load bundle處理類(select for broker)：org.apache.pulsar.broker.loadbalance.impl.LeastLongTermMessageRate

如上圖，假設三個 Broker 平均負載是 50%，則閾值就是 60%，超出 60% 的部分需要均衡。但實際應用中發現 Broker 1 的多餘 20% 負載會解除安裝到 Broker 2 上，之後由於 Broker 2 超載所以又會解除安裝下來，還會回到 Broker 1 上。結果流量就在 Broker 1 和 Broker 2 上反覆橫跳。

跟蹤程式碼發現，Load Bundle 處理類是根據 Broker 的訊息量判斷該承載多餘流量的 Broker，但生產中訊息量與機器負載並不完全正相關，且 Threshold shedder 是根據 CPU、出入流量、記憶體等多種指標平均加權得出 Broker 負載，所以 bundle 的載入和解除安裝邏輯並不一致。

對此團隊進行了程式碼優化改進：

loadManagerClassName=org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl
loadBalancerLoadSheddingStrategy=org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder
loadBalancerBrokerThresholdShedderPercentage=10
loadBalancerBrokerOverloadedThresholdPercentage=70
Load bundle處理類(select for broker)：在低於平均負載的broker中隨機選擇
loadBalancerDistributeBundlesEvenlyEnabled=false (相同的程式碼實現：PR-16059)

優化後的效果如下，可以看到叢集流量穩定許多：

團隊還在實時推薦場景下優化了 Broker 快取。這種場景有以下特徵：

消費任務數量眾多；
消費速度參差不齊；
消費任務經常重啟。

對此，社群原有的 Broker 快取邏輯效果不佳。以下是 Broker 快取的原有驅逐邏輯：

void doCacheEviction(long maxTimestamp) {
    if (entryCache.getsize() &lt;= 0) {
        return;
    }
    // Always remove all entries already read by active cursors
    PositionImpl slowestReaderPos = getEarlierReadPositionForActiveCursors);
    if (slowestReaderPos != null) {
        entryCache.invalidateEntries(slowestReaderPos):
    }
    // Remove entries older than the cutoff threshold
    entryCache.invalidateEntriesBeforeTimestamp(maxTimestamp);
}

預設策略會找出當前消費不活躍（由閾值控制，Cursor 消費的 entry 超過閾值即被認為是不活躍）的 Cursor，對 Cursor 之前的資料做驅逐。對此，騰訊工程師向社群提交了程式碼改進：

void doCacheEviction (long maxTimestamp){
    if (entryCache.getSize() (= 0) {
        return;
    )
    PositionImpl evictionPos;
    if (config.isCacheEvictionByMarkDeletedPosition()){
        evictionPos=getEarlierMarkDeletedPositionForActiveCursors().getNext();
    } else {
        // Always remove all entries already read by active cursors
        evictionPos=getEarlierReadPositionForActiveCursors();
    }
    if (evictionPos != null) {
        entryCache.invalidateEntries(evictionPos);
    }
    // Remove entries older than the cutoff threshold
    entryCache.invalidateEntriesBeforeTimestamp(maxTimestamp);
}

這裡將選擇非活躍 Cursor 的邏輯改成了尋找需要刪除的資料位置。這樣消費速度相對較慢的資料就不會穿越到 Bookie 中增加叢集壓力，只要資料有 Backlog 就會被快取。但這種方法會導致快取空間吃緊，因為消費任務重啟期間仍舊要無意義地保留快取，佔用快取空間。

對此微信團隊在社群改進的基礎上又做了調整：

void doCacheEviction(long maxTimestamp){
    if (entryCache.getSize() &lt;= 0) {
        return;
    }
    if (factory.getConfig().isRemoveReadEntriesInCache()){
        PositionImpl evictionPos;
        if (config.isCacheEvictionByMarkDeletedPosition()){
            PositionImplearlierMarkDeletedPosition=getEarlierMarkDeletedPositionForActiveCursors();
            evictionPos = earlierMarkDeletedPosition != null? earlierMarkDeletedPosition.getNext() : null;
        } else {
            // Always remove all entries already read by active cursors
            evictionPos=getEarlierReadPositionForActiveCursors();
        }
        if (evictionPos != null) {
            entryCache.invalidateEntries(evictionPos);
        }
    }
    //Remove entries older than the cutoff threshold
    entryCache.invalidateEntriesBeforeTimestamp(maxTimestamp);
}

這裡簡單地將一定時間內的資料快取到 Broker 中，有效提升了場景中的快取效率：

實踐 4：COS Offloader 開發與應用

Pulsar 提供了分層儲存能力，可以將儲存轉移到廉價的儲存層。Pulsar Offloader 可以將超過一定時長的 Ledger 搬運到遠端儲存，不再停留在 Bookie 層，由 Broker 接管這部分的資料管理。

團隊使用 Pulsar Offloader 的原因有：

Bookie Journal/Ledger 盤都使用 SSD，成本較高；
業務需求儲存時間長、資料儲存量大；
資料消費任務異常，需要容忍較長時間的資料 Backlog；
資料回放需求。

Pulsar 社群版本並不支援騰訊雲物件儲存（COS），所以團隊開發了內部雲上 COS Offloader 外掛並應用於線上。

未來展望與計劃

團隊在部署與使用過程中一直和社群密切溝通，團隊未來計劃跟進社群版本升級與 bug 修復。微信團隊將著重參與一些特性，比如 PIP 192 Broker 負載均衡與快取優化，計劃重構負載均衡器；PIP 180 通過影子 Topic 解決讀放大問題，幫助精細化管理 Topic。微信團隊也在關注 Pulsar 生態進展，如 Flink、Pulsar、資料湖全鏈路打通。

以上就是Apache Pulsar 微信大流量實時推薦場景下實踐詳解的詳細內容，更多關於Apache Pulsar微信大流量推薦的資料請關注it145.com其它相關文章！