Pipeline這個(gè)詞做開發(fā)的、做數(shù)據(jù)的、做算法的幾乎天天見但真被問到“Pipeline到底是什么、怎么把它吃透”很多人又容易犯迷糊。就拿我自己來說早年剛接觸這個(gè)概念時(shí)以為就是“把幾個(gè)腳本串起來跑”后來踩了不少坑才慢慢摸清楚它背后那一整套設(shè)計(jì)邏輯——從一個(gè)輸入到最終產(chǎn)出結(jié)果中間經(jīng)歷采集、清洗、計(jì)算、存儲(chǔ)、展示等各個(gè)環(huán)節(jié)每個(gè)環(huán)節(jié)都可能成為瓶頸每個(gè)環(huán)節(jié)都需要琢磨可靠性、容錯(cuò)、性能、血緣、可觀測(cè)性。這篇就用一個(gè)具體的案例把Pipeline從設(shè)計(jì)到落地到調(diào)優(yōu)的整個(gè)路徑完整跑一遍順便把我在實(shí)操中踩過的雷和總結(jié)出來的經(jīng)驗(yàn)一起分享出來。開門見山我選了一個(gè)很典型的場(chǎng)景來做演示一個(gè)跨平臺(tái)的“用戶行為分析流水線”。這個(gè)案例足夠有代表性因?yàn)樗烊话瑪?shù)據(jù)采集、消息緩沖、實(shí)時(shí)計(jì)算、離線批處理、結(jié)果存儲(chǔ)、可視化展示多個(gè)環(huán)節(jié)幾乎覆蓋了日常開發(fā)和運(yùn)維中會(huì)和Pipeline打交道的所有痛點(diǎn)。你可以把它當(dāng)成一個(gè)縮影往后不管是做推薦系統(tǒng)特征管道、音視頻處理鏈路、電商訂單流轉(zhuǎn)還是模型訓(xùn)練流水線都能從這個(gè)案例里找到可以遷移的思考方式。這個(gè)案例要解決的實(shí)際問題很明確某產(chǎn)品需要實(shí)時(shí)統(tǒng)計(jì)用戶在不同頁(yè)面上的點(diǎn)擊、曝光和轉(zhuǎn)化行為同時(shí)每天還要產(chǎn)出全量報(bào)表。一開始團(tuán)隊(duì)直接用數(shù)據(jù)庫(kù)輪詢加腳本算數(shù)據(jù)量過了千萬級(jí)之后集群經(jīng)常被打掛報(bào)表出得又慢實(shí)時(shí)指標(biāo)對(duì)不上。折騰了幾輪之后我們決定把這套邏輯完整地重構(gòu)成一條Pipeline并且把“每一個(gè)環(huán)節(jié)都要能獨(dú)立解釋、獨(dú)立運(yùn)維、獨(dú)立優(yōu)化”作為最核心的設(shè)計(jì)標(biāo)準(zhǔn)。這篇文章不是純理論科普也不是教你怎么用某個(gè)具體開源框架而是從需求和設(shè)計(jì)思路出發(fā)把選型、實(shí)現(xiàn)、測(cè)試、排查、優(yōu)化這幾個(gè)環(huán)節(jié)完整串一遍。適合誰看正在搭建數(shù)據(jù)處理鏈路的開發(fā)者和架構(gòu)師需要理解Pipeline設(shè)計(jì)理念的算法工程師以及身邊總有人把“Pipeline”掛在嘴上但不知道該怎么落地的初學(xué)者。1. Pipeline整體設(shè)計(jì)與階段拆解1.1 一個(gè)Pipeline到底在拆什么一談到Pipeline我腦子里首先出現(xiàn)的不是代碼而是工廠里的流水線。原料從一頭進(jìn)去經(jīng)歷不同工位逐步加工成成品。每個(gè)工位只負(fù)責(zé)一件事工位之間用傳送帶連接傳送帶的快慢和緩存能力決定了整條線能不能平穩(wěn)運(yùn)轉(zhuǎn)?；氐竭@個(gè)用戶行為分析案例我們實(shí)際要處理的是下面這堆工序數(shù)據(jù)采集在App、小程序、Web端埋點(diǎn)把用戶點(diǎn)擊、頁(yè)面瀏覽、登錄、下單這些行為事件發(fā)到后端接口這一步是整個(gè)Pipeline的源頭源頭如果斷了后面全部沒有意義。消息緩沖采集服務(wù)收到的行為事件不能立刻隨手寫數(shù)據(jù)庫(kù)而是先推到消息隊(duì)列里起到削峰填谷的作用。高峰期的流量可能是平時(shí)的幾十倍如果沒有緩沖數(shù)據(jù)庫(kù)瞬間就會(huì)被寫入壓力擊垮。實(shí)時(shí)計(jì)算從消息隊(duì)列消費(fèi)數(shù)據(jù)做清洗和聚合把用戶的行為在秒級(jí)內(nèi)變成可用的統(tǒng)計(jì)結(jié)果比如“當(dāng)前在線人數(shù)”“過去5分鐘的點(diǎn)擊量”。離線批處理每天固定時(shí)間把當(dāng)天的原始事件撈出來重新做一次全量計(jì)算生成精確的日?qǐng)?bào)表和實(shí)時(shí)結(jié)果相互校驗(yàn)。結(jié)果存儲(chǔ)與查詢把實(shí)時(shí)聚合結(jié)果和離線報(bào)表結(jié)果分別落到適合查詢的存儲(chǔ)中按業(yè)務(wù)需求提供查詢接口?？梢暬故景汛鎯?chǔ)的結(jié)果以看板的形式呈現(xiàn)給運(yùn)營(yíng)和管理人員。每條Pipeline的環(huán)節(jié)不一樣但本質(zhì)上都是這種“多階段接力”的結(jié)構(gòu)。每個(gè)階段的輸入是上一個(gè)階段的輸出輸出又成為下一個(gè)階段的輸入。這種結(jié)構(gòu)最大的好處在于可以在單個(gè)階段上做優(yōu)化和替換而不用像單體腳本那樣“動(dòng)一處就要全量回歸”。1.2 為什么是消息隊(duì)列作“傳送帶”流水線里最容易被忽略但又最關(guān)鍵的東西就是工位之間的傳送帶。在數(shù)據(jù)處理這個(gè)語境下這個(gè)傳送帶一般就是消息隊(duì)列。選消息隊(duì)列而不是直接用HTTP接口把數(shù)據(jù)從采集端直傳計(jì)算端核心原因有三個(gè)第一是解耦。采集端不需要關(guān)心下游是誰、下游是否在線只需要把消息投遞給隊(duì)列就算完成計(jì)算端也不需要關(guān)心上游來了多少數(shù)據(jù)、什么時(shí)候來只管按自己的速度消費(fèi)。生產(chǎn)和消費(fèi)之間完全不感知對(duì)方的存在這在系統(tǒng)變更頻繁的初期非常重要。第二是削峰。行為數(shù)據(jù)有明顯的波峰波谷比如晚上八點(diǎn)可能每分鐘涌入幾百萬條事件凌晨三點(diǎn)可能每分鐘只有幾千條。如果直接打給計(jì)算服務(wù)計(jì)算服務(wù)必須按最高峰值來擴(kuò)容否則就丟數(shù)據(jù)。有了消息隊(duì)列生產(chǎn)者可以一直全速寫入消費(fèi)者在高峰期也可以暫時(shí)堆積消息避免了資源浪費(fèi)和系統(tǒng)崩潰。第三是可靠性。消息寫入消息隊(duì)列后隊(duì)列本身會(huì)負(fù)責(zé)持久化和副本復(fù)制輕易不會(huì)丟失。消費(fèi)者掛掉之后可以從上次消費(fèi)的位置繼續(xù)不需要依賴業(yè)務(wù)系統(tǒng)自己實(shí)現(xiàn)復(fù)雜的重試和恢復(fù)邏輯。在這個(gè)案例中我們最終選了分區(qū)順序語義的消息中間件作為核心傳送帶因?yàn)樾袨閿?shù)據(jù)分析對(duì)時(shí)間順序有一定要求。同一用戶的行為如果亂序會(huì)影響路徑歸因和漏斗分析的結(jié)果。既然費(fèi)用和運(yùn)維復(fù)雜度都在可控范圍內(nèi)帶來的收益卻非常明確這筆投入相當(dāng)劃算。1.3 實(shí)時(shí)與離線“雙軌制”的設(shè)計(jì)考量有人會(huì)問既然做了實(shí)時(shí)計(jì)算為什么還要另外跑離線批處理重復(fù)建設(shè)一套這個(gè)問題的答案隱藏在兩種計(jì)算模式各自的優(yōu)劣里。實(shí)時(shí)計(jì)算采用的是近似計(jì)算邏輯為了滿足低延遲它會(huì)在內(nèi)存里做窗口聚合一旦窗口翻轉(zhuǎn)聚合結(jié)果就固化輸出事后不會(huì)重新計(jì)算。這帶來一個(gè)讓人揪心的問題如果統(tǒng)計(jì)口徑后來改了或者窗口邊界存在幾秒鐘的誤差實(shí)時(shí)結(jié)果就沒辦法修正。離線批處理不一樣它面對(duì)的是全量歷史數(shù)據(jù)可以慢慢算允許重跑結(jié)果可復(fù)現(xiàn)。哪怕今天發(fā)現(xiàn)昨天的報(bào)表算錯(cuò)了把昨天的數(shù)據(jù)重新?lián)瞥鰜碓倥芤槐轳R上就能得到正確結(jié)果。我們的設(shè)計(jì)策略是“雙軌并行、以離線為基準(zhǔn)、以實(shí)時(shí)為準(zhǔn)繩”。實(shí)時(shí)結(jié)果支撐對(duì)時(shí)效性敏感的頁(yè)面展示離線結(jié)果用于每天的正式報(bào)表和深入分析。兩邊算出同一個(gè)指標(biāo)后如果差距在允許范圍內(nèi)說明鏈路一切正常如果差異過大就要回頭查實(shí)時(shí)鏈路哪里消費(fèi)慢了、窗口丟數(shù)據(jù)了、或者消息積壓沒處理完。這套雙軌制帶來的額外收益是讓線上和離線互相成為“測(cè)試用例”。很多時(shí)候數(shù)據(jù)Schema變更了、埋點(diǎn)字段改名了實(shí)時(shí)鏈路沒暴露問題離線跑批反而會(huì)報(bào)錯(cuò)這就是最好的預(yù)警機(jī)制。2. 核心組件選型與關(guān)鍵技術(shù)點(diǎn)解析2.1 采集端設(shè)計(jì)埋點(diǎn)協(xié)議和傳輸方式埋點(diǎn)是整條Pipeline的起點(diǎn)也是最容易埋雷的地方。我們?cè)诎咐惺褂昧艘惶捉y(tǒng)一的事件協(xié)議把行為劃分為幾大類曝光事件、點(diǎn)擊事件、頁(yè)面停留事件、業(yè)務(wù)轉(zhuǎn)化事件。每一類事件都帶公共字段包括用戶ID、會(huì)話ID、時(shí)間戳、頁(yè)面標(biāo)識(shí)、客戶端類型再加上各事件特有的業(yè)務(wù)字段。這個(gè)協(xié)議看起來簡(jiǎn)單但對(duì)后續(xù)所有環(huán)節(jié)影響深遠(yuǎn)。比如用戶ID的生成邏輯必須統(tǒng)一一旦有的端用設(shè)備ID、有的端用賬號(hào)ID后面做用戶去重和留存分析時(shí)就要花大量時(shí)間做ID映射。時(shí)間戳也必須統(tǒng)一我們約定所有端上報(bào)的時(shí)間戳必須使用客戶端事件發(fā)生時(shí)間而不是服務(wù)器接收時(shí)間。原因很簡(jiǎn)單客戶端在弱網(wǎng)環(huán)境會(huì)延遲上報(bào)服務(wù)器接收時(shí)間只能代表“到的時(shí)間”不能代表“發(fā)生的時(shí)間”如果不統(tǒng)一這一點(diǎn)后續(xù)所有時(shí)序統(tǒng)計(jì)全部會(huì)失真。傳輸方式上我們選擇了批量異步上報(bào)?？蛻舳讼劝咽录彺嬖诒镜?cái)€夠一批或者每隔幾秒發(fā)送一次而不是每發(fā)生一個(gè)事件就立刻發(fā)一個(gè)HTTP請(qǐng)求。這樣既能減少網(wǎng)絡(luò)開銷又能規(guī)避用戶快速滑頁(yè)面時(shí)產(chǎn)生大量并發(fā)請(qǐng)求打爆網(wǎng)關(guān)的問題。這里有個(gè)細(xì)節(jié)值得注意客戶端的緩沖隊(duì)列可能會(huì)丟數(shù)據(jù)。所以我們?cè)赟DK中做了持久化緩存和文件重放機(jī)制App殺進(jìn)程后恢復(fù)啟動(dòng)會(huì)把上次沒發(fā)出去的事件補(bǔ)發(fā)。2.2 實(shí)時(shí)計(jì)算引擎的選擇邏輯實(shí)時(shí)計(jì)算環(huán)節(jié)我們對(duì)比了兩種主流方案純手寫消費(fèi)者做流式處理以及使用流處理框架。純手寫消費(fèi)者最大的優(yōu)勢(shì)是靈活可以精確控制消費(fèi)邏輯、內(nèi)存占用、并發(fā)模型和依賴庫(kù)幾十上百行代碼就能跑起來一個(gè)簡(jiǎn)單應(yīng)用。缺點(diǎn)是所有容錯(cuò)、狀態(tài)管理、窗口計(jì)算、重試機(jī)制都要自己實(shí)現(xiàn)維護(hù)成本非常高。如果只是測(cè)試環(huán)境驗(yàn)證用手寫沒問題但若核心指標(biāo)都依賴這條鏈路手寫方案的工程債早晚會(huì)壓垮團(tuán)隊(duì)。最終我們選了成熟流處理框架理由也很務(wù)實(shí)框架內(nèi)置消息消費(fèi)位點(diǎn)管理、分布式快照、窗口聚合、狀態(tài)存儲(chǔ)、故障恢復(fù)和背壓處理這些能力這些功能如果全部自研隨便一個(gè)可能都要寫幾個(gè)月的代碼。流處理框架基于事件時(shí)間和處理時(shí)間都有完善支持正好匹配我們“客戶端事件時(shí)間優(yōu)先”的設(shè)計(jì)。有一點(diǎn)我覺得需要特別提醒引入框架不等于萬事大吉。框架只是把編程模型變簡(jiǎn)單了底層資源、數(shù)據(jù)傾斜、算子并發(fā)度、內(nèi)存分配這些仍然要自己規(guī)劃和調(diào)優(yōu)。就像自動(dòng)擋汽車確實(shí)不用踩離合了但油路、輪胎、剎車系統(tǒng)還是要定期檢查和維護(hù)的。2.3 離線批處理怎么和“數(shù)據(jù)湖”接軌離線批處理鏈路我們采用了經(jīng)典的分層思路。原始數(shù)據(jù)落地到一個(gè)分布式存儲(chǔ)文件系統(tǒng)上的統(tǒng)一數(shù)據(jù)目錄按日期和小時(shí)分區(qū)。這一層被我們稱為“原始層”它的定位是只做追加、不做修改、不刪數(shù)據(jù)。任何人都可以訪問這層數(shù)據(jù)做臨時(shí)探查但誰都不允許回頭改寫歷史。再往上是“明細(xì)層”把原始事件經(jīng)過清洗、過濾、解析、字段補(bǔ)齊之后按照業(yè)務(wù)邏輯重新組織。比如把不同端上報(bào)的異構(gòu)事件統(tǒng)一成標(biāo)準(zhǔn)Schema把用戶ID解析成統(tǒng)一標(biāo)識(shí)剔除機(jī)器人流量造成的刷量數(shù)據(jù)。最上面才是“應(yīng)用層”根據(jù)業(yè)務(wù)指標(biāo)定義做預(yù)聚合生成報(bào)表數(shù)據(jù)。這一層直接對(duì)接數(shù)據(jù)可視化系統(tǒng)查詢速度要求很高所以通常以寬表和指標(biāo)表的方式存儲(chǔ)。這種分層設(shè)計(jì)的好處是隔離和復(fù)用。原始層保存的是“事實(shí)”明細(xì)層保存的是“清洗后的真相”應(yīng)用層保存的是“特定口徑的結(jié)果”。臨時(shí)跑一個(gè)分析需求只需要在最合適的層級(jí)上動(dòng)手不會(huì)影響已有的報(bào)表。2.4 存儲(chǔ)選型不同查詢模式需要不同的數(shù)據(jù)庫(kù)存儲(chǔ)環(huán)節(jié)很容易犯一個(gè)錯(cuò)誤試圖用一個(gè)數(shù)據(jù)庫(kù)包打天下。我們發(fā)現(xiàn)實(shí)時(shí)查詢、明細(xì)查詢和報(bào)表查詢的模式差異非常大分開存儲(chǔ)才是更合理的選擇。實(shí)時(shí)聚合結(jié)果數(shù)據(jù)量不大但更新頻繁查詢側(cè)要求毫秒級(jí)返回所以用速度快的KV型引擎按時(shí)間和維度聚合鍵作為主鍵保存指標(biāo)值。明細(xì)數(shù)據(jù)查詢通常是按用戶ID或者按時(shí)間范圍做掃描查詢頻率低、數(shù)據(jù)量大更適合存儲(chǔ)和分析分離的OLAP引擎用列式存儲(chǔ)加索引來支持快速聚合。報(bào)表數(shù)據(jù)是高度預(yù)聚合的寬表每天固定寫入一些行和列查詢模式非常固定放在支持標(biāo)準(zhǔn)SQL的關(guān)系型數(shù)據(jù)庫(kù)反而最合適運(yùn)營(yíng)直接拖個(gè)Excel工具就能連上去查交互上對(duì)業(yè)務(wù)人員比較友好。三種存儲(chǔ)職責(zé)分明各自承擔(dān)壓力最大的那部分查詢整體運(yùn)維復(fù)雜度雖然增加了但收益是任何一條鏈路出現(xiàn)抖動(dòng)都不會(huì)波及其他鏈路。3. 實(shí)操過程從零搭建整個(gè)Pipeline3.1 環(huán)境準(zhǔn)備與基礎(chǔ)依賴實(shí)操開始之前先把環(huán)境準(zhǔn)備好。我們的測(cè)試環(huán)境一共用了若干臺(tái)服務(wù)器節(jié)點(diǎn)其中一臺(tái)作為管理節(jié)點(diǎn)兩臺(tái)作為工作節(jié)點(diǎn)另外單獨(dú)部署了消息隊(duì)列集群和存儲(chǔ)引擎集群。操作系統(tǒng)的調(diào)整不多主要是調(diào)大文件句柄數(shù)上限、關(guān)閉內(nèi)存過度分配以及設(shè)置時(shí)間同步這幾項(xiàng)是消息隊(duì)列集群和分布式計(jì)算框架運(yùn)行的硬性要求。代碼層面我們統(tǒng)一用一套獨(dú)立開發(fā)的SDK管理數(shù)據(jù)接入?yún)f(xié)議。所有端都打包同一個(gè)協(xié)議庫(kù)協(xié)議字段的增刪改都通過這套SDK發(fā)版避免不同端各自拷貝一份接口定義變成一改就崩的局面。項(xiàng)目結(jié)構(gòu)上整個(gè)倉(cāng)庫(kù)做了多模塊劃分采集端服務(wù)、消息生產(chǎn)者、流式計(jì)算任務(wù)、批處理任務(wù)、查詢API服務(wù)分別位于獨(dú)立目錄模塊之間通過接口定義交互。提示時(shí)間同步這里千萬別跳過。分布式環(huán)境里節(jié)點(diǎn)間系統(tǒng)時(shí)間偏差超過幾百毫秒消息隊(duì)列的時(shí)序就會(huì)亂數(shù)據(jù)寫入后按時(shí)間戳排序的結(jié)果也會(huì)錯(cuò)。生產(chǎn)環(huán)境踩過這個(gè)坑排查了很久才發(fā)現(xiàn)是某個(gè)節(jié)點(diǎn)的時(shí)鐘漂移導(dǎo)致的。3.2 采集模塊的配置文件示例采集端服務(wù)接收客戶端上報(bào)行為事件校驗(yàn)協(xié)議之后寫入消息隊(duì)列。配置如下# 采集服務(wù)配置 server.port8080 server.accept.threads16 server.max.pending.requests4096 # 消息隊(duì)列配置 mq.bootstrap.serversnode01:9092,node02:9092,node03:9092 mq.topic.behavioruser-behavior-events mq.producer.acks1 mq.producer.retries3 mq.producer.batch.size16384 mq.producer.linger.ms50這里簡(jiǎn)單解釋幾個(gè)關(guān)鍵參數(shù)的設(shè)置邏輯。acks1表示生產(chǎn)者寫入消息隊(duì)列主節(jié)點(diǎn)即返回確認(rèn)兼顧了吞吐量和可靠性。如果要求不丟消息可以配置acksall但延遲會(huì)明顯升高。我們分析行為數(shù)據(jù)允許極端情況下極少量消息丟失所以acks1在成本和可靠性之間找到了一個(gè)很好的平衡點(diǎn)。linger.ms50和batch.size16384在實(shí)際效果上有協(xié)同關(guān)系。生產(chǎn)者沒有立刻發(fā)送每條消息而是在本地?cái)€了最多50毫秒積攢到16KB大小后批量發(fā)出。這樣做的收益是請(qǐng)求次數(shù)大幅減少網(wǎng)絡(luò)往返開銷大幅降低消息隊(duì)列的寫入TPS可以成倍提升。代價(jià)是消息延遲增加了最多50毫秒對(duì)于行為分析場(chǎng)景完全可接受。3.3 流式計(jì)算任務(wù)的窗口策略流式計(jì)算任務(wù)消費(fèi)消息隊(duì)列中的行為事件核心邏輯是按時(shí)間窗口做聚合統(tǒng)計(jì)。我們需求里有兩個(gè)核心實(shí)時(shí)指標(biāo)每分鐘各頁(yè)面的點(diǎn)擊次數(shù)以及每五分鐘各頁(yè)面的獨(dú)立訪客數(shù)。這種場(chǎng)景天然適合滾動(dòng)窗口。窗口策略配置如下作業(yè)名稱user-behavior-streaming-job 時(shí)間語義事件時(shí)間從事件字段中提取客戶端時(shí)間戳 水位線策略允許5秒亂序延遲 窗口類型滾動(dòng)窗口窗口大小分別為60秒和300秒 輸出模式窗口結(jié)束時(shí)輸出一次聚合結(jié)果這里最值得展開的是“水位線”這個(gè)概念。它是事件時(shí)間處理中的核心機(jī)制本質(zhì)是一個(gè)時(shí)間標(biāo)記只有當(dāng)水位線越過某個(gè)時(shí)間點(diǎn)系統(tǒng)才認(rèn)為該時(shí)間點(diǎn)之前的事件都已經(jīng)到達(dá)可以關(guān)閉對(duì)應(yīng)的窗口并觸發(fā)計(jì)算。因?yàn)榫W(wǎng)絡(luò)延遲、客戶端設(shè)備差異和數(shù)據(jù)處理耗時(shí)的存在事件到達(dá)流處理引擎的順序往往和事件發(fā)生的順序不一致如果不處理亂序窗口統(tǒng)計(jì)就會(huì)偏低。我們?cè)试S5秒的亂序延遲意味著水位線被設(shè)定為當(dāng)前觀察到的事件時(shí)間戳再減去5秒。晚到不超過5秒的事件仍然能進(jìn)入正確窗口參與計(jì)算。代價(jià)是每個(gè)窗口的結(jié)果會(huì)比自然時(shí)間晚5秒左右輸出這對(duì)實(shí)時(shí)看板來說完全無感。注意這個(gè)參數(shù)不是越大越好。亂序延遲設(shè)置得大會(huì)導(dǎo)致結(jié)果嚴(yán)重滯后設(shè)置得太小又可能丟失大量亂序數(shù)據(jù)。合理的做法是先觀察真實(shí)線上數(shù)據(jù)的亂序分布曲線取P95附近的延遲值作為水位線參數(shù)。3.4 離線批處理任務(wù)的實(shí)現(xiàn)邏輯離線批處理任務(wù)用標(biāo)準(zhǔn)SQL腳本實(shí)現(xiàn)每天凌晨定時(shí)調(diào)度執(zhí)行。整個(gè)腳本分三層第一層從原始數(shù)據(jù)目錄讀取昨天的全部事件CREATE TABLE IF NOT EXISTS ods_user_event AS SELECT * FROM external_table WHERE dt ${yesterday}第二層清洗和標(biāo)準(zhǔn)化。這一步要把采集到的字段統(tǒng)一格式過濾掉明顯異常的測(cè)試流量補(bǔ)齊IP歸屬地等維度然后寫入明細(xì)層表。第三層按業(yè)務(wù)指標(biāo)做聚合。比如獨(dú)立訪客數(shù)的精確計(jì)算因?yàn)樯婕皟|級(jí)規(guī)模去重我們當(dāng)時(shí)采用了兩種方式并行驗(yàn)證一種是精確去重用大內(nèi)存計(jì)算保證準(zhǔn)確另一種是近似去重追求速度和資源消耗均衡。日常報(bào)表用精確值臨時(shí)探查用近似值兩者相差在一定范圍內(nèi)就說明數(shù)據(jù)質(zhì)量沒大問題。3.5 消息積壓的觀察與控制手段消息積壓是流式Pipeline最經(jīng)典、最容易出現(xiàn)的問題之一。消費(fèi)速度跟不上生產(chǎn)速度消息在隊(duì)列里越堆越多實(shí)時(shí)報(bào)表的延遲從一個(gè)窗口漲到幾個(gè)小時(shí)最終從“實(shí)時(shí)”變成“次日更新”。我們當(dāng)時(shí)排查積壓?jiǎn)栴}的步驟很簡(jiǎn)單先通過消息隊(duì)列的消費(fèi)組狀態(tài)命令查看每個(gè)分區(qū)的消費(fèi)位點(diǎn)和最新位點(diǎn)之間的差值。確定哪些分區(qū)積壓嚴(yán)重后再看消費(fèi)端每個(gè)分區(qū)分配到的數(shù)據(jù)量是否均衡。最后看流處理作業(yè)中每個(gè)算子的繁忙率定位瓶頸算子。有一次積壓的根因非常隱蔽。某個(gè)維表關(guān)聯(lián)操作在算子內(nèi)部實(shí)現(xiàn)了同步調(diào)用遠(yuǎn)程數(shù)據(jù)庫(kù)的邏輯每次關(guān)聯(lián)請(qǐng)求需要幾十毫秒到幾百毫秒。單看延遲感覺還好但在每秒幾萬條數(shù)據(jù)的高流量下這部分同步等待把整體吞吐拖到了原來的五分之一。改成異步維表查詢之后吞吐立即恢復(fù)了。這個(gè)經(jīng)驗(yàn)說明在流處理鏈路里最怕的不是計(jì)算復(fù)雜度而是阻塞等待。4. 性能測(cè)試與調(diào)優(yōu)實(shí)錄4.1 壓測(cè)方案設(shè)計(jì)與參數(shù)選擇Pipeline搭建完成后不能直接上線要先做一輪壓測(cè)摸清系統(tǒng)的真實(shí)水位。我們的壓測(cè)方案分三塊數(shù)據(jù)模擬、場(chǎng)景設(shè)計(jì)、觀測(cè)指標(biāo)。數(shù)據(jù)模擬方面寫了一個(gè)事件生成器按照真實(shí)用戶的訪問分布模擬不同類型事件的產(chǎn)生概率并注入一定的亂序和延遲力求接近線上數(shù)據(jù)特征。壓測(cè)場(chǎng)景分三個(gè)梯度平時(shí)流量的兩倍、峰值流量的五倍、以及突發(fā)極端流量的十倍。每一輪持續(xù)三十分鐘前十分鐘讓流量逐步爬到目標(biāo)水位中間十五分鐘保持穩(wěn)定最后五分鐘允許流量自然回落。這種階梯式設(shè)計(jì)可以準(zhǔn)確觀察不同流量級(jí)別下Pipeline各環(huán)節(jié)的延遲和吞吐變化。觀測(cè)指標(biāo)不只盯著“每秒處理了多少條”這一個(gè)數(shù)字。我們同時(shí)記錄四個(gè)維度消息隊(duì)列生產(chǎn)速率、消息隊(duì)列消費(fèi)速率、窗口計(jì)算延遲和結(jié)果存儲(chǔ)寫入時(shí)間。四個(gè)指標(biāo)一起看才能發(fā)現(xiàn)瓶頸到底在哪一段。4.2 性能瓶頸定位與處理措施壓測(cè)過程中發(fā)現(xiàn)了一個(gè)很典型的瓶頸消息隊(duì)列消費(fèi)端的單分區(qū)處理能力成為掣肘。具體表現(xiàn)是隨著發(fā)送速率增大消息生產(chǎn)者一切正常消費(fèi)者卻頻繁出現(xiàn)拉取超時(shí)。問題原因在于我們對(duì)事件進(jìn)行了按用戶ID的哈希分區(qū)目的是保證同一用戶的數(shù)據(jù)落在同一分區(qū)中。但如果事件總量集中在少數(shù)幾個(gè)高活躍用戶身上這些用戶對(duì)應(yīng)的分區(qū)就會(huì)嚴(yán)重過載而其他分區(qū)卻很空閑。我們把這個(gè)稱為“熱點(diǎn)分區(qū)”。我們做了一次針對(duì)性優(yōu)化默認(rèn)按用戶ID哈希分區(qū)保持同一用戶序列完整同時(shí)針對(duì)高活躍用戶建立更細(xì)粒度的二級(jí)分流機(jī)制將其部分事件單獨(dú)分流到冷分區(qū)。這個(gè)改動(dòng)上線后整體吞吐提升了約40%消息積壓的隱患解除了。4.3 內(nèi)存參數(shù)調(diào)整與GC優(yōu)化流處理作業(yè)跑在Java虛擬機(jī)里大狀態(tài)、大窗口都對(duì)內(nèi)存極為敏感。我們一開始使用默認(rèn)垃圾回收配置運(yùn)行平穩(wěn)之后在高峰期頻繁出現(xiàn)stop-the-world表現(xiàn)為窗口計(jì)算結(jié)果延遲波動(dòng)極大。解決思路分三步。第一步調(diào)大堆內(nèi)存同時(shí)把堆內(nèi)外內(nèi)存比例規(guī)劃好避免對(duì)象頻繁在堆內(nèi)外來回拷貝。第二步換用更適合大堆場(chǎng)景的垃圾回收器把GC暫停時(shí)間從幾次長(zhǎng)暫停變成多次短暫停實(shí)時(shí)性有了肉眼可見的改善。第三步分析對(duì)象生命周期的分布把頻繁創(chuàng)建的短生命周期對(duì)象盡量改成復(fù)用減少Young區(qū)壓力。這三步做完整個(gè)作業(yè)在全量高峰數(shù)據(jù)下依然能保持穩(wěn)定的輸出間隔。調(diào)優(yōu)的方向很多但沒有必要一次全上。每次只動(dòng)一個(gè)變量觀察效果確認(rèn)之后再動(dòng)下一個(gè)。否則出了問題根本不知道是哪次改動(dòng)引起的。4.4 端到端延遲的觀測(cè)方法Pipeline是由多個(gè)獨(dú)立子系統(tǒng)拼接而成的每個(gè)子系統(tǒng)都有自己的日志和監(jiān)控但“端到端的延遲”才是用戶真正感知到的指標(biāo)。需要一套統(tǒng)一追蹤機(jī)制把各環(huán)節(jié)串聯(lián)起來。我們的做法是給每個(gè)事件分配一個(gè)全局唯一的追蹤ID。采集端生成事件時(shí)把追蹤ID作為字段寫入消息消息隊(duì)列層通過消息頭傳播流處理算子把追蹤ID透?jìng)鞯骄酆辖Y(jié)果里最終查詢結(jié)果時(shí)可以根據(jù)追蹤ID反查這個(gè)結(jié)果到底經(jīng)歷了哪些節(jié)點(diǎn)、每個(gè)節(jié)點(diǎn)耗時(shí)多少。這套機(jī)制上線后的第一個(gè)收獲是發(fā)現(xiàn)大部分端到端延遲并非來自計(jì)算環(huán)節(jié)而是來自消息隊(duì)列中的批量等待。默認(rèn)配置下生產(chǎn)者為了湊夠批量大小會(huì)刻意等待更多消息這在低峰期會(huì)額外增加幾十毫秒延遲。針對(duì)實(shí)時(shí)性要求最高的場(chǎng)景我們單獨(dú)配置了更短的等待時(shí)間延遲立刻降了下來。5. 數(shù)據(jù)一致性保障與鏈路可靠性5.1 冪等寫入是Pipeline的基礎(chǔ)要求Pipeline里最容易出現(xiàn)的問題之一是數(shù)據(jù)的重復(fù)或者丟失。消息隊(duì)列用于解耦削峰時(shí)本身不保證“恰好一次”投遞語義——它可能正好一次也可能重復(fù)消費(fèi)者一旦在處理過程中掛掉重啟就會(huì)出現(xiàn)部分消息沒處理完就被重新消費(fèi)的情況。為了保證最終結(jié)果準(zhǔn)確最常用的手段是讓整個(gè)鏈路支持冪等寫入。所謂冪等就是“同一條數(shù)據(jù)無論被寫入多少次最終的結(jié)果和只寫入一次是一樣的”。在實(shí)時(shí)聚合場(chǎng)景我們用帶主鍵的存儲(chǔ)表來保證冪等。每個(gè)聚合結(jié)果的主鍵由“窗口開始時(shí)間 聚合維度鍵”組成。消息重復(fù)消費(fèi)時(shí)生成的主鍵相同寫入操作變成對(duì)同一條記錄的覆蓋不會(huì)產(chǎn)生重復(fù)數(shù)據(jù)。在離線批處理場(chǎng)景我們用分區(qū)覆蓋和主鍵合并來實(shí)現(xiàn)冪等。每次寫入明細(xì)層時(shí)先清理目標(biāo)分區(qū)再寫入本次計(jì)算的數(shù)據(jù)。即使任務(wù)重跑結(jié)果也是確定的?？吹降膯栴}就是任何環(huán)節(jié)如果接受“重復(fù)一次沒關(guān)系”這種偷懶的思路早晚會(huì)在某次鏈路上游重放時(shí)爆炸。5.2 死信機(jī)制與異常事件處理即使做了完整的校驗(yàn)和過濾依然會(huì)有少量消息內(nèi)容異常比如字段缺失、時(shí)間戳不合法、嵌套JSON解析失敗。流處理框架遇到這類畸形數(shù)據(jù)時(shí)如果直接忽略問題會(huì)被悄然掩埋如果直接報(bào)錯(cuò)整個(gè)任務(wù)又會(huì)因一條壞數(shù)據(jù)中斷更不可取。我們的做法是接入死信隊(duì)列機(jī)制。解析失敗的原始消息經(jīng)過格式化處理之后被完整地投遞到一個(gè)專門的異常事件主題中。這個(gè)主題里保存的每條消息都附帶了失敗原因和原始內(nèi)容。日常運(yùn)維中只要看到死信主題有內(nèi)容就說明上游埋點(diǎn)或者傳輸鏈路有不大不小的質(zhì)量問題可以及時(shí)推送給數(shù)據(jù)負(fù)責(zé)人去處理而不是等到最終報(bào)表對(duì)不上才發(fā)現(xiàn)。死信主題還有第二個(gè)妙用可以做數(shù)據(jù)回放。當(dāng)上游修復(fù)了埋點(diǎn)邏輯之后死信主題里的歷史異常消息可以重新觸發(fā)一次清洗讓這部分?jǐn)?shù)據(jù)恢復(fù)納入統(tǒng)計(jì)。這個(gè)能力在關(guān)鍵報(bào)表審計(jì)時(shí)真的救過我們一次。5.3 數(shù)據(jù)質(zhì)量規(guī)則引擎的建立Pipeline的穩(wěn)定性不能只靠人盯著監(jiān)控。我們建立了一套數(shù)據(jù)質(zhì)量規(guī)則引擎讓機(jī)器每天自動(dòng)檢查數(shù)據(jù)是否滿足既定的質(zhì)量卡點(diǎn)。規(guī)則大致分三類。第一類是完整性規(guī)則例如“核心事件類型的每日總量波動(dòng)和前一天相比不應(yīng)該超過±30%”。如果某個(gè)事件量驟降60%基本不是業(yè)務(wù)變化導(dǎo)致的而是埋點(diǎn)消失、采集服務(wù)出故障或者傳輸鏈路被攔截。第二類是時(shí)效性規(guī)則例如“當(dāng)天事件數(shù)據(jù)的產(chǎn)出時(shí)間不應(yīng)該晚于次日凌晨六點(diǎn)”。一旦離線任務(wù)延遲超過閾值說明資源不足或者某個(gè)環(huán)節(jié)已經(jīng)阻塞很久需要告警觸發(fā)擴(kuò)容。第三類是口徑一致性規(guī)則例如“實(shí)時(shí)報(bào)表的當(dāng)日累計(jì)值和離線日?qǐng)?bào)表的核心指標(biāo)誤差不應(yīng)超過±5%”。雙軌制在這里發(fā)揮價(jià)值一旦誤差超標(biāo)系統(tǒng)自動(dòng)發(fā)告警。這個(gè)即時(shí)預(yù)警能力遠(yuǎn)比事后人工核對(duì)可靠得多。這套規(guī)則引擎并不是一次建完就結(jié)束的而是持續(xù)積累的過程。每發(fā)現(xiàn)一類新問題沉淀出一條新規(guī)則規(guī)則庫(kù)越來越厚Pipeline本身也就越來越強(qiáng)壯。6. 常見問題、排查方法與實(shí)踐經(jīng)驗(yàn)總結(jié)6.1 消息亂序引發(fā)的統(tǒng)計(jì)偏差排查亂序問題要按從上游到下游的順序逐層確認(rèn)先看客戶端上報(bào)時(shí)間是否為事件發(fā)生時(shí)間再看消息隊(duì)列中消息的寫入和讀取順序最后看流處理引擎的事件時(shí)間處理和水位線是否配置正確。每次只觀察一個(gè)環(huán)節(jié)的日志和監(jiān)控定位會(huì)快得多。我們有一陣子發(fā)現(xiàn)某個(gè)頁(yè)面的點(diǎn)擊漏斗轉(zhuǎn)化率總是異常后來定位到原因是新版本升級(jí)后客戶端在弱網(wǎng)下批量重傳了舊事件舊事件被重放后覆蓋了正常事件的時(shí)序統(tǒng)計(jì)。解決辦法是在事件體中加入“上報(bào)批次號(hào)”和“本地發(fā)送序號(hào)”在流處理端做去重和亂序丟棄。6.2 晚到的數(shù)據(jù)要不要補(bǔ)算這是所有做實(shí)時(shí)數(shù)據(jù)的人都糾結(jié)過的問題。晚到的數(shù)據(jù)如果一律補(bǔ)算窗口結(jié)果會(huì)反復(fù)變化報(bào)表不穩(wěn)定如果一律丟棄又會(huì)低估真實(shí)指標(biāo)。我們的原則很實(shí)用根據(jù)數(shù)據(jù)對(duì)賬口徑來決定。正式對(duì)賬以離線批處理結(jié)果為準(zhǔn)實(shí)時(shí)數(shù)據(jù)給出的數(shù)字是參考性質(zhì)所以實(shí)時(shí)鏈路可以容忍一定程度的晚到數(shù)據(jù)缺失。針對(duì)需要精確對(duì)賬的核心指標(biāo)我們單獨(dú)設(shè)置更長(zhǎng)亂序容忍窗口確保大部分常規(guī)延遲數(shù)據(jù)都能漸進(jìn)補(bǔ)入。絕大多數(shù)場(chǎng)景下這種“實(shí)時(shí)給快值、離線出真值”的分工比死磕“每一條都要實(shí)時(shí)計(jì)算完全準(zhǔn)確”要靠譜得多。6.3 組件升級(jí)引發(fā)的連鎖故障Pipeline里任何組件的升級(jí)都不是“換一個(gè)版本”那么簡(jiǎn)單必須做全鏈路兼容性驗(yàn)證。我們經(jīng)歷過一次升級(jí)消息隊(duì)列客戶端后新版本客戶端默認(rèn)開啟了一個(gè)壓縮算法消息隊(duì)列服務(wù)端版本較舊解壓時(shí)直接拋出異常導(dǎo)致整個(gè)鏈路消費(fèi)中斷超過十分鐘。從那以后我們定下了一條規(guī)矩所有連接組件的版本升級(jí)都要在預(yù)發(fā)環(huán)境先跑一段線上流量影子確認(rèn)無誤后再灰度發(fā)布。升級(jí)消息隊(duì)列時(shí)先升級(jí)消息隊(duì)列服務(wù)端再升級(jí)生產(chǎn)者客戶端最后升級(jí)消費(fèi)者客戶端嚴(yán)格保持順序和兼容性。6.4 獨(dú)家避坑經(jīng)驗(yàn)總結(jié)做Pipeline的時(shí)間越長(zhǎng)越覺得很多坑其實(shí)是相通的。先總結(jié)幾條所有做數(shù)據(jù)鏈路相關(guān)的人都值得記住的經(jīng)驗(yàn)。鏈路必須先有監(jiān)控再談優(yōu)化。沒有prometheus式的埋點(diǎn)沒有每一跳的耗時(shí)、積壓和成功率數(shù)據(jù)所謂優(yōu)化就是盲人摸象。監(jiān)控先行不是流程上的儀式而是Pipeline工程的基本盤。任何時(shí)候不要丟掉原始數(shù)據(jù)。存儲(chǔ)成本再高也要能回溯到最原始的事件記錄。沒有原始數(shù)據(jù)的Pipeline就像沒有底賬的財(cái)務(wù)系統(tǒng)出了任何問題都無從追責(zé)和修復(fù)?！笆褂贸墒炜蚣堋辈坏扔凇伴]眼不用管”?？蚣茇?fù)責(zé)抽象掉通用問題但由于業(yè)務(wù)參數(shù)、數(shù)據(jù)特征、集群規(guī)模帶來的差異化問題仍然必須自己負(fù)責(zé)。框架只是工具不是大腦。你的Pipeline要能“解釋自己”。任何一個(gè)環(huán)節(jié)的輸入是什么、輸出是什么、延遲多少、積壓多少、錯(cuò)誤率多少都應(yīng)該能以數(shù)字的形式展示出來。做不到這點(diǎn)的Pipeline隨著規(guī)模增大總有一天會(huì)變成所有人都害怕碰的怪獸。6.5 用這套思路去應(yīng)對(duì)其他Pipeline場(chǎng)景文章里這個(gè)用戶行為分析案例只是一個(gè)起點(diǎn)。如果把這些經(jīng)驗(yàn)遷移到別的場(chǎng)景你會(huì)發(fā)現(xiàn)底層邏輯是相通的。做推薦系統(tǒng)的特征Pipeline時(shí)最重要的不是特征計(jì)算速度而是特征一致性和回放能力。離線訓(xùn)練時(shí)生成的特征和在線推理時(shí)實(shí)時(shí)拼出來的特征必須完全一致否則線上模型效果永遠(yuǎn)達(dá)不到離線實(shí)驗(yàn)的水平。這個(gè)問題的解法就是在Pipeline里始終堅(jiān)持“同一套特征邏輯、同一份特征代碼”的原則。做音視頻處理Pipeline時(shí)最核心的是每一條任務(wù)的可重試性和斷點(diǎn)續(xù)傳能力。轉(zhuǎn)碼、抽幀、合成這些操作往往耗時(shí)很長(zhǎng)任何一個(gè)下游環(huán)節(jié)失敗如果必須從第一個(gè)環(huán)節(jié)重新開始成本是無法忍受的。所以需要在任務(wù)分解和進(jìn)度持久化上多下功夫。做模型訓(xùn)練Pipeline時(shí)最容易被忽視的是數(shù)據(jù)版本和特征版本的管理。模型訓(xùn)練出了效果要能回溯到底用的是哪份數(shù)據(jù)、哪套特征、哪版代碼否則迭代和復(fù)現(xiàn)流程會(huì)寸步難行。不管做什么領(lǐng)域Pipeline工程的內(nèi)核是相通的清晰的階段邊界、完善的可靠性機(jī)制、透明的觀測(cè)能力、可控的排錯(cuò)手段、以及保證數(shù)據(jù)一致性的底線原則。最后再分享一個(gè)小技巧定期對(duì)整個(gè)Pipeline做一次“停電演練”把每一個(gè)階段主動(dòng)停掉一段時(shí)間看恢復(fù)后整個(gè)系統(tǒng)能否自動(dòng)回到一致狀態(tài)。這個(gè)演練表面上是在給自己找麻煩實(shí)際上是用最小的代價(jià)提前暴露了系統(tǒng)在真實(shí)故障時(shí)可能出現(xiàn)的問題。做過幾次演練之后你會(huì)發(fā)現(xiàn)要對(duì)一條Pipeline建立信心靠的不是運(yùn)氣而是這種“敢于主動(dòng)制造故障、再自信修復(fù)”的反復(fù)打磨。