七牛是如何搞定每天500億條日志的 轉

七牛是如何搞定每天500億條日志的

牛小七2015年7月31日釋出在 技術分享

7月30日，七牛資料平台工程師王團結在CSDN Spark微信使用者群，與近千名Spark技術開發人員，結合七牛内部使用的資料平台，深入分享了團隊是如何利用Flume、Kafka、Spark Streaming等技術搞定每天500億條日志的，并詳細講解了各個工具使用的注意點。王團結，主要負責七牛資料平台的設計研發工作，關注大資料處理和 高性能系統服務，對Hadoop、Flume、Kafka、Spark等離線、分布式計算技術十分感興趣。以下為演講實錄。

概述

資料平台在大部分公司都屬于支撐性平台，做的不好立刻會被吐槽，這點和運維部門很 像。是以在技術選型上優先考慮現成的工具，快速出成果，沒必要去擔心有技術負擔。早期，我們走過彎路，認為沒多少工作量，收集存儲和計算都自己研發，發現 是吃力不讨好。去年上半年開始，我們全面擁抱開源工具，搭建自己的資料平台。

公司的主要資料來源是散落在各個業務伺服器上的半結構化日志，比如系統日志、程式日 志、通路日志、審計日志等。日志是最原始的資料記錄，如果不是日志，肯定會有資訊上的丢失。說個簡單的例子，需求是統計Nginx上每個域名的的流量，這 個完全可以通過一個簡單的Nginx子產品去完成，但是如果需要統計不同來源的流量就無法做了，是以需要原始的完整的日志。

有種手法是業務程式把日志通過網絡直接發送出去，但是這并不可取，因為網絡和接收端并不完全可靠，當出問題時會對業務造成影響或者日志丢失。是以，對業務侵入最小最自然的方式是把日志落到本地硬碟上。

資料平台設計架構

Agent設計需求

每台機器上會有一個Agent去同步這些日志，這是個典型的隊列模型，業務程序在不 斷的push，Agent在不停地pop。Agent需要有記憶功能，用來儲存同步的位置（offset），這樣才盡可能保證資料準确性，但不可能做到完 全準确。由于發送資料和儲存offset是兩個動作，不具有事務性，不可避免的會出現資料不一緻性情況，通常是發送成功後儲存offset，那麼在 Agent異常退出或機器斷電時可能會造成多餘的資料。 Agent需要足夠輕，這主要展現在運維和邏輯兩個方面。Agent在每台機器上都會部署，運維成本、接入成本是需要考慮的。Agent不應該有解析日 志、過濾、統計等動作，這些邏輯應該給資料消費者。倘若Agent有較多的邏輯，那它是不可完成的，不可避免的經常會有更新變更動作。

資料收集流程

資料收集這塊的技術選擇，Agent是用Go自己研發的，消息中間件Kafka，數 據傳輸工具Flume。說到資料收集經常有人拿Flume和Kafka做比較，我看來這兩者定位是不同的，Flume更傾向于資料傳輸本身，Kakfa是 典型的消息中間件用于解耦生産者消費者。

具體架構上，Agent并沒把資料直接發送到Kafka，在Kafka前面有層由Flume構成的forward。這樣做有兩個原因：

• Kafka的API對非JVM系的語言支援很不友好，forward對外提供更加通用的HTTP接口。
• forward層可以做路由、Kafka topic和Kafka partition key等邏輯，進一步減少Agent端的邏輯。

forward層不含狀态，完全可以做到水準擴充，不用擔心成為瓶頸。出于高可用考 慮，forward通常不止一個執行個體，這會帶來日志順序問題，Agent按一定規則（round-robin、failover等）來選擇forward 執行個體，即使Kafka partition key一樣，由于forward層的存在，最終落入Kafka的資料順序和Agent發送的順序可能會不一樣。我們對亂序是容忍的，因為産生日志的業務基 本是分布式的，保證單台機器的日志順序意義不大。如果業務對順序性有要求，那得把資料直接發到Kafka，并選擇好partition key，Kafka隻能保證partition級的順序性。

多機房的情形，通過上述流程，先把資料彙到本地機房Kafka叢集，然後彙聚到核心機房的Kafka，最終供消費者使用。由于Kafka的mirror對網絡不友好，這裡我們選擇更加的簡單的Flume去完成跨機房的資料傳送。

Flume使用要點

Flume在不同的資料源傳輸資料還是比較靈活的，但有以下幾個點需要注意。

• memory-channel效率高但可能有丢資料的風險，file- channel安全性高但性能不高。我們是用memory-channel，但把capacity設定的足夠小，使記憶體中的資料盡可能少，在意外重新開機和斷 電時丢的資料很少。個人比較排斥file-channel，效率是一方面，另一個是對Flume的期望是資料傳輸，引入file-channel時，它的 角色會向存儲轉變，這在整個流程中是不合适的。通常Flume的sink端是Kafka和HDFS這種可用性和擴張性比較好的系統，不用擔心資料擁堵問 題。
• 預設的HTTP source沒有設定線程池，有性能問題，如果有用到，需要修改代碼。
• 單sink速度跟不上時，需要多個sink。像跨機房資料傳輸網絡延遲高單rpc sink吞吐上不去和HDFS sink效率不高情形，我們在一個channel後會配十多個sink。

Kafka使用要點

Kafka在性能和擴充性很不錯，以下幾個點需要注意下。

• topic的劃分，大topic對生産者有利且維護成本低，小topic對消費者比較友好。如果是完全不相關的相關資料源且topic數不是發散的，優先考慮分topic。
• Kafka的并行機關是partition，partition數目直接關系整體的吞吐量，但parition數并不是越大越高，3個partition就能吃滿一塊普通硬碟IO了。是以partition數是由資料規模決定，最終還是需要硬碟來抗。
• partition key選擇不當，可能會造成資料傾斜。在對資料有順序性要求才需使用partition key。Kafka的producer sdk在沒指定partition key時，在一定時間内隻會往一個partition寫資料，這種情況下當producer數少于partition數也會造成資料傾斜，可以提高 producer數目來解決這個問題。

資料離線和實時計算

資料到Kafka後，一路資料同步到HDFS，用于離線統計，另一路用于實時計算。由于今天時間有限，接下來隻能和大家分享下實時計算的一些經驗。

實時計算選擇的是Spark Streaming。目前隻有統計需求，沒疊代計算的需求，Spark Streaming使用比較保守。從Kakfa讀資料統計完存入MongoDB中，中間狀态資料很少，好處是系統吞吐量很大，但沒遇到記憶體相關問題。

Spark Streaming對儲存計算結果的資料庫TPS要求較高。假如有10w個域名需要統計流量，batch interval為10s，每個域名有4個相關統計項，算下來平均是4w TPS，這隻是平均值，實際峰值更高，固态硬碟上的MongoDB也隻能抗1w TPS，後續我們會考慮用Redis來抗這麼高的TPS。

當開啟speculation參數或代碼層面沒處理好異常時，task可能會被重放。但是有外部狀态的task是不可重入的，否則會造成計算結果的不準确。說個簡單的例子，如下：

這個任務，如果被重放了，會造成落入MongoDB的結果比實際多。

有些對象會包含狀态，這些狀态的生成需要較大的代價，不能做到在每次使用時都去new一個。我們對這種對象的處理政策是JVM内一個對象，同時在代碼層面做好并發控制。類似下面：

如果這種使用形式有性能問題，可以考慮實作一個pool來管理。

在Spark1.3的後版本，streaming為Kafka引入了新的 Direct API試圖解決資料準确性問題。Direct API把Kafka consumer offset的管理暴露出來（舊版本是異步存入Zookeeper），讓使用者透明的管理consumer offset，這在一定程度上能緩解準确性問題，但不可避免還會有一緻性問題。為什麼這樣說呢？隻有計算結果和offset兩者的儲存具有事務性，才能完 全準确。這個事務有兩種手段做到，一種是用Mysql這種支援事務的資料庫，把計算結果和offset的儲存放在一個事務裡，另一種是自己實作兩階段提 交。Direct API 還會有性能問題，因為它到計算的時候才實際從Kafka讀資料，這對整體吞吐有很大影響。

七牛資料平台規模

要分享的就這些了，最後秀下實時計算這邊規模：Flume＋Kafka＋Spark 混部在8台高配機器，日均500億條資料，峰值80w TPS。

轉載于:https://www.cnblogs.com/gym333/p/4711170.html