一、背景

支付中心作為攜程集團公共部門，主要負(fù)責(zé)的業(yè)務(wù)包括交易、實名綁卡、賬戶、收單等，由于涉及到交易相關(guān)的資金流轉(zhuǎn)以及用戶實名認(rèn)證，部分用戶操作環(huán)節(jié)的中間數(shù)據(jù)應(yīng)內(nèi)控/審計要求需要長時間保存。當(dāng)前研發(fā)應(yīng)用多，日志量大、格式各異，對于日志的存儲和使用產(chǎn)生較大的挑戰(zhàn)，故支付數(shù)據(jù)與研發(fā)團隊群策群力，共同開發(fā)了一套統(tǒng)一日志框架。

二、總體架構(gòu)圖 

核心模塊包括：日志生產(chǎn)、日志采集、日志解析，其中調(diào)用流程如下：

• 研發(fā)應(yīng)用/服務(wù)接入基于log4j2擴展的統(tǒng)一日志組件，將日志拋送至kafka。
• 周期性啟動消費kafka topic的camus job將日志寫入hdfs。
• T+1啟動MR job讀取camus寫入的hdfs內(nèi)容并load到hive表。

三、日志生產(chǎn)-統(tǒng)一日志組件 

支付研發(fā)基于log4j2自定義了多個Appender，將應(yīng)用日志以服務(wù)調(diào)用形式拋送至kafka，并被log_process_service 服務(wù)統(tǒng)一處理并提交至攜程常用基礎(chǔ)日志框架如：CLOG、CAT、ES，各應(yīng)用無需關(guān)心公司日志框架，統(tǒng)一由日志平臺處理。

其優(yōu)點：

• 不同日志框架對應(yīng)著不同的Appender，方便進行日志框架接入的擴展。
• 采用AOP編程，對于接入的業(yè)務(wù)侵入性小，接入簡單。
• 定義了豐富的java注解，便于日志配置化輸出，其中可打印日志包括但不限于：類名、方法名、方法入?yún)ⅰ⒎祷刂?、異常等，支持敏感字段脫敏?/li>

存在的問題：

• 日志格式不規(guī)范：研發(fā)應(yīng)用數(shù)百個，研發(fā)人員較多，日志格式差異大，給數(shù)據(jù)分析和使用帶來巨大挑戰(zhàn)。
• 存儲時長短：當(dāng)前公司在線CLOG存儲系統(tǒng)只能查詢最近幾天數(shù)據(jù)、ES保存稍長一段時間數(shù)據(jù)且不支持批量查詢，基礎(chǔ)離線CLOG hive表由于數(shù)據(jù)量巨大，僅能做到T+2，無法滿足T+1的報表需求。

故支付數(shù)據(jù)團隊在研發(fā)團隊統(tǒng)一日志組件的基礎(chǔ)上，結(jié)合數(shù)據(jù)分析和數(shù)據(jù)存儲生命周期開發(fā)了統(tǒng)一日志框架。

3.1 統(tǒng)一日志-埋點設(shè)計

支付研發(fā)團隊負(fù)責(zé)數(shù)百個服務(wù)或應(yīng)用，支持的業(yè)務(wù)包括：路由、鑒權(quán)、免密、卡服務(wù)、訂單、錢包實名、電子支付等，不同的業(yè)務(wù)又可拆分app、h5、online、offline等項目，整合這些數(shù)據(jù)是個極大的挑戰(zhàn)。如果各系統(tǒng)研發(fā)埋點任意指定，會給BI數(shù)據(jù)分析帶來極大的困難，數(shù)據(jù)分析準(zhǔn)確性難以得到保障，故支付數(shù)據(jù)基于業(yè)務(wù)特點定義了一套統(tǒng)一日志埋點規(guī)范。

字段定義主要是基于日常分析需求，致力于簡化數(shù)據(jù)的使用，故總體原則為json形式，當(dāng)前原始日志有兩部分組成：tag/message，其中tag數(shù)據(jù)結(jié)構(gòu)為Map，關(guān)鍵數(shù)據(jù)一般是通過tag內(nèi)的數(shù)據(jù)進行檢索，明細(xì)數(shù)據(jù)通過message進行檢索，tag與message的組成格式為：[[$tag]]$message，目前標(biāo)準(zhǔn)字段包括兩類：規(guī)范性字段和通用性字段。

3.1.1 規(guī)范性字段格式

規(guī)范性字段需要應(yīng)用研發(fā)一定程度的參與，提供符合字段命名的類和方法。部分字段名稱及定義如下：

• 字段名稱字段類型描述serviceNamestring調(diào)用服務(wù)名稱tagMapkeyvalue信息messagestring原始日志requeststring接口請求參數(shù)responsestring接口返回值requesttimestring日志請求時間responsetimestring日志請求時間

其中tag可以靈活填充，主要擴展字段如下：

• 名稱字段類型描述versionstringapp版本號platstring平臺信息refnostring流水號

3.1.2 通用字段格式

• 日志框架能夠自動獲取屬性，無需研發(fā)編碼，即可打印。
• 字段名稱字段類型描述applicationidstring攜程應(yīng)用唯一識別號logtimestring日志生成時間

3.2 分區(qū)/分桶字段的定義

當(dāng)前離線數(shù)據(jù)分析基于hive引擎，hive的分區(qū)分桶設(shè)計極大的影響了查詢性能，特別是在日志量巨大的場景下，分區(qū)字段的選擇尤為關(guān)鍵。如：用戶進入支付收銀臺可能會有上百個場景，而每種場景下會有多次服務(wù)調(diào)用，其中不同場景下服務(wù)調(diào)用頻率差異很大，占用的空間差異也較大，故針對每種場景分配一個唯一的場景號，通過場景號進行分區(qū)，可以高效的進行數(shù)據(jù)分析，而過多的分區(qū)也可能導(dǎo)致較多的小文件，對hadoop namenode產(chǎn)生較大的影響，此時結(jié)合分桶能夠達到優(yōu)化查詢效率且避免分區(qū)無限制野蠻增長產(chǎn)生眾多過多小文件的問題。

四、日志采集 

日志采集框架基于LinkedIn的開源項目Camus，Camus使用MapReduce讀取kafka數(shù)據(jù)然后寫入hdfs，由于無reduce階端，所有數(shù)據(jù)處理及寫入都在Map側(cè)，很少會發(fā)生數(shù)據(jù)傾斜，Camus具有接入簡單，方便擴展，故我們進行了二次開發(fā)，以滿足當(dāng)前業(yè)務(wù)的需要：

• 自定義decoder/partitioner，原生的decoder/partitioner支持的hdfs落地路徑主要基于日期，較為粗糙，無法適應(yīng)業(yè)務(wù)的需要。故自定義decoder 抽取原始日志分區(qū)字段，然后代入partitioner中，生成具有業(yè)務(wù)含義的hdfs輸出路徑，為特定時間范圍數(shù)據(jù)回刷提供了高效的解決方案。
• 自定義provider，原生的StringRecordWriterProver僅支持text文件方式落地，占用空間大、壓縮后無法并行切分，容易錯列錯行，而orc格式數(shù)據(jù)，有效的節(jié)約了hdfs占用空間，查詢效率高且可以切分，有利于日志解析job的高效執(zhí)行。其中在配置Camus job過程中需要關(guān)注如下問題：

4.1 camus 任務(wù)執(zhí)行

執(zhí)行頻率設(shè)置

Error: java.io.IOException: target exists.the file size(614490 vs 616553) is not the same. 

由于kafka消息保存天數(shù)有限和單個分區(qū)size有限(Server 配置：log.retention.bytes)，攜程側(cè)為3天和10G，如果camus同步kafka頻率較低時，可能會出現(xiàn)數(shù)據(jù)丟失，故需要根據(jù)日志量大小，設(shè)置camus 調(diào)度任務(wù)的執(zhí)行頻率，防止數(shù)據(jù)丟失。

任務(wù)重疊執(zhí)行

Error: java.io.IOException: target exists.the file size(614490 vs 616553) is not the same. 

camus從kafka 讀取數(shù)據(jù)，任務(wù)要以單例形式執(zhí)行，任務(wù)執(zhí)行完成后才會更新kafka的offset，若一個任務(wù)執(zhí)行了多次，就會導(dǎo)致數(shù)據(jù)大小無法對齊，此時需要刪除配置路徑下的所有數(shù)據(jù)后重新啟動任務(wù)，即可完成修復(fù)。

4.2 如何控制camus落地文件的大小

當(dāng)kafka各partition數(shù)據(jù)寫入量不平衡時，由于各partition會寫入一個hdfs文件中，如果研發(fā)日志集中寫入kafka某個partition，會導(dǎo)致這個partition對應(yīng)的hdfs文件占用空間特別大，如果恰巧這個文件是不可切分的，極端情況下會導(dǎo)致只有一個線程去解析這個大文件，降低了數(shù)據(jù)讀寫的并發(fā)度，拉長了數(shù)據(jù)解析時間，遇到這種問題的解決辦法是：

• 臨時解決方案：研發(fā)日志分散寫入kafka partition，不要導(dǎo)致某類數(shù)據(jù)集中寫入一個partition;
• 高效解決方案：數(shù)據(jù)側(cè)采用可切分的輸入格式，進行數(shù)據(jù)切分;

4.3 寫入orc文件格式注意事項

orc寫入timeout

AttemptID:attempt_1587545556983_2611216_m_000001_0 Timed out after 600 secs 

orc文件寫入速度較text文件會慢很多，如果同時寫入的的文件較多或者內(nèi)存回收占用時間較長，會導(dǎo)致map方法在600秒內(nèi)沒有讀、寫或狀態(tài)更新，job會被嘗試終結(jié)，解決方法是調(diào)高默認(rèn)的task超時時間，由10分鐘調(diào)高到20分鐘。

mapreduce.task.timeout=1200000 

OOM 內(nèi)存溢出

beyond physical memory limits. Current usage: 2.5 GB of 2.5 GB physical memory used; 4.2 GB of 5.3 GB virtual memory used. Killing container. 

在orc寫文件的時候如果出行較多的OOM，此時需要加大map執(zhí)行的內(nèi)存。

mapreduce.map.memory.mb=8096 
mapreduce.map.java.opts=-Xmx6000m 

五、統(tǒng)一日志-解析

鑒于日志解析工作主要集中在MapReduce的Map側(cè)，而Map側(cè)通過參數(shù)調(diào)整能夠很容易控制map的個數(shù)，以提高數(shù)據(jù)解析的并發(fā)度，MapReduce主要分為：intputformat、map、shuffle、reduce等幾個核心階段，通過優(yōu)化各階段的執(zhí)行時間，可以顯著提高日志解析的速度。

5.1 inputsplit優(yōu)化

MR job中影響map的個數(shù)主要有：

• 文件個數(shù)：如果不采用CombineFileInputFormat，那么不會進行小文件合并，每個文件至少有一個map處理，當(dāng)小文件太多時，頻繁啟動和回收線程也會對性能產(chǎn)生影響，同時對集群其它job資源分配產(chǎn)生影響。
• 文件屬性：當(dāng)文件較大且可切分時，系統(tǒng)會生成多個map處理大文件，inputsplit塊按照MR最小單元進行文件切割(split)，并且一個split對應(yīng)一個MapTask。

前期日志解析程序的性能較高，一天的全量日志解析約25分鐘，中間有段時間任務(wù)執(zhí)行時間從25分鐘延遲到4個小時，原因是研發(fā)將大量訂單號為空的日志寫入到指定的partition中，日志量巨大，導(dǎo)致其中少量map在讀取大文件時執(zhí)行時間特別長。

經(jīng)過分析發(fā)現(xiàn)text+snappy 文件無法切分，只能夠被一個map處理，將camus落地數(shù)據(jù)格式從text+snappy換為orc+snappy格式，同時開發(fā)了支持orc文件格式的CombineFileInputFormat，既減少了小文件對hadoop計算資源果斷的占用也提高了job的并發(fā)程度。

5.2 shuffle優(yōu)化

使map的輸出能夠更加均勻的映射到reduce側(cè)，由于默認(rèn)的分區(qū)策略是對map的輸出key hash取reduce個數(shù)的模，容易導(dǎo)致數(shù)據(jù)傾斜，解決辦法是在key上面增加時間戳或者重寫partition函數(shù)。

5.3 批量日志解析

當(dāng)前MR的輸出會作為hive外表的數(shù)據(jù)源，hive表會按照業(yè)務(wù)過程進行分區(qū)，所有數(shù)據(jù)的解析結(jié)果路徑為：日期+業(yè)務(wù)過程，而業(yè)務(wù)過程可能有數(shù)百個，采用了MultipleInputs/MultipleOutputs 能夠在一個mapreduce job中實現(xiàn)多輸入多輸出的功能，以適應(yīng)業(yè)務(wù)自定義解析，并歸一化后統(tǒng)一拋到reduce側(cè)。

5.3.1 空文件生產(chǎn)

在使用的過程中會出現(xiàn)生成眾多臨時小文件及生成size 為0的小文件，增加了hdfs namenode內(nèi)存壓力，同時空文件也會導(dǎo)致spark表查詢失敗，可通過LazyOutputFormat進行修復(fù)。

5.3.2 文件重復(fù)創(chuàng)建

MultipleOutputs輸出文件一般以name-r-nnnnn的格式進行命名，其中name與程序指定的文件名有關(guān)，nnnnn表示reduce任務(wù)號。在處理數(shù)據(jù)較多時，可能會存在reduce側(cè)反復(fù)創(chuàng)建已存在的文件，導(dǎo)致任務(wù)長時間運行而不能成功，中間生成了大量小文件，對hadoop namenode產(chǎn)生較大壓力，影響整個集群響應(yīng)時間。

解決方案為：在reduce側(cè)進行數(shù)據(jù)寫入時，需要對exception進行捕捉，一旦出現(xiàn)數(shù)據(jù)寫入exception，即將對應(yīng)的寫入reduce文件刪除并終止程序，由于MR支持高可用，當(dāng)一個reduce taks 失敗后會自動重試，重試一定次數(shù)依然不能夠成功就會導(dǎo)致整個任務(wù)失敗，每次重試避免了不停的重復(fù)創(chuàng)建已存在的文件，引起NN響應(yīng)時間極速下降。 

5.4 reduce個數(shù)調(diào)整

目前日志解析的reduce側(cè)主要用于orc數(shù)據(jù)寫入，當(dāng)reduce個數(shù)較少時，會導(dǎo)致reduce內(nèi)存溢出，而reduce個數(shù)較多時，可能會導(dǎo)致非常多的小文件且占用集群過多資源，可以通過計算map側(cè)輸入文件的個數(shù)及總占用空間，動態(tài)計算需要的reduce個數(shù)，以達到合理利用資源的目的。

六、日志治理

日志落地導(dǎo)致的一個問題是存儲空間增長迅速，當(dāng)前支付中心日均新增ORC壓縮原始數(shù)據(jù)量TB級別且還在持續(xù)增長中。

支付數(shù)據(jù)側(cè)根據(jù)研發(fā)、產(chǎn)品的需求對不同類型日志進行分級，對于不同類別的日志設(shè)置不同的存儲周期，主要劃分為：研發(fā)排障日志、審計日志、數(shù)據(jù)分析日志等;同時在camus將日志寫入hdfs時，由于按照業(yè)務(wù)分區(qū)進行落地，導(dǎo)致生成了大量小文件，需要對這些小文件進行合并并且設(shè)置TTL，避免對hadoop namenode產(chǎn)生較大的影響。

七、總結(jié)與展望

目前日均TB級數(shù)據(jù)解析時間在30分鐘內(nèi)完成，后期計劃將日志系統(tǒng)導(dǎo)入clickhouse等對實時要求高的場景共運營使用，以支持業(yè)務(wù)精細(xì)化運營和分析。

【作者簡介】英明，攜程數(shù)據(jù)研發(fā)專家，負(fù)責(zé)支付離線數(shù)據(jù)倉庫建設(shè)及BI業(yè)務(wù)需求，對并行計算、大數(shù)據(jù)處理及建模等有濃厚興趣。