摘要：關于友盟數據架構友盟架構思想友盟的架構主要參考了提出的架構思想。關于友盟實踐通過以上的介紹，大家可能對整個大數據平臺的結構和概念有了初步的了解。所以接下來，我給大家分享一些友盟在實踐中得到的一些經驗。友盟的數據平臺經歷了一個發展的過程。

摘要：友盟大數據平臺的架構借鑒了Lambda架構思想，數據接入層讓Kafka集群承擔，后面由Storm消費，存儲在MongoDB里面，通過Kafka自帶的Mirror功能同步，兩個Kafka集群，可以分離負載；計算有離線和實時兩部分，實時是Storm，離線是Hadoop，數據倉庫用Hive，數據挖掘正在從Pig遷移到Spark，大量的數據通過計算之后，存儲在HDFS上，最后存儲在HBase里面，通過ES來提供多級索引，以彌補HBase二級索引的缺失......

友盟從 2010 年成立開始就專注移動大數據， 5 年來不僅積累了大量的數據，而且積累了豐富的技術和經驗，那么友盟的大數據平臺是怎樣架構的呢？在剛剛結束的MDCC 2015 大會上，友盟的數據平臺負責人吳磊在 Android 專場為大家做了技術分享。

以下是吳磊的口述，整理時略有刪減。

關于友盟大數據平臺

目前，友盟合作的 App 超過 64 萬 ，同時為 23 萬開發者團隊提供服務，截止到2015年7月底，友盟數據平臺總量 9 PB，每天增量壓縮后有 7TB，每天要處理接近 82 億的對話，實時處理 100K QPS，離線處理 800 多個常規任務，集群規模是 500 多臺服務器， 14000 個 CPU 核心。

關于友盟數據架構

友盟架構思想

友盟的架構主要參考了Twitter提出的Lambda架構思想。

如上圖所示，最下面是快速處理層，新增數據在快速處理層計算，這部分數據比較小，可以快速完成，生成實時視圖。

同時，新增數據會并入全量數據集，進行批處理，生成批處理視圖。這樣，系統同時具有了低延遲實時處理能力，也具有離線大數據處理能力。之后通過數據服務層，把兩個視圖合并起來，對外提供服務。在外部看來，這是一個完整的視圖。 就這樣，通過 lambda 架構我們可以將實時處理和批處理系統巧妙的結合起來。

友盟數據平臺整體架構

根據友盟的業務特點，數據平臺由下向上分成這幾個部分：最基礎的是日志收集，接下來進入離線計算和實時分析，計算后的結果，會進行數據挖掘，有價值的數據進入數據倉庫。接下來友盟會提供一個基于 REST Service的數據服務，在此服務之上做各種數據應用例如：報表、數據分析報告，數據下載等等。兩邊的部分提供輔助的功能：包括任務調度和監控管理。

友盟數據流水線

結合友盟的業務架構和 lambda 架構思想，最終的系統如下圖所示：最左邊是數據采集層，友盟提供手機、平板、盒子的SDK給App集成，App通過SDK發送日志到友盟平臺；首先進入到Nginx，負載均衡之后傳給基于finagle框架的日志接收器，接著來到數據接入層。

友盟使用兩個 Kafka 集群來承擔數據接入功能。上面這個Kafka集群被實時計算消費。下面的kafka是用于離線數據消費，兩個集群之間通過Kafka的mirror功能進行同步。這么做的主要目的是IO負載的分離，避免離線部分過大的IO請求影響到實時計算部分；以及實時離線部分的業務解耦，方便兩部分業務獨立發展。

接下來是數據計算層，分別由離線計算層和實時計算層組成。離線部分，我們主要是基于Hadoop Mapreduce框架開發了一系列的MR任務。同時使用Hive建立我們的數據倉庫，使用pig進行數據挖掘，現在我們也在逐步使用Spark來承擔數據挖掘相關的工作。實時部分是通過storm來進行流式計算。

實時部分的計算結果會存儲到 MongoDB，離線部分的數據存儲在 HDFS 。離線分析的結果存儲在 HBase 。因為 HBase 缺少二級索引的相關功能，所以我們引入了 Elastic Search 來為 HBase 相關表提供索引查詢功能。最后通過統一的 REST Service 來對外提供數據服務。

關于友盟實踐

通過以上的介紹，大家可能對整個大數據平臺的結構和概念有了初步的了解。正如Linux 之父的名言，“Talk is cheap, show me the code！”一樣，其實知道是相對容易的，難的是如何去實現。所以接下來，我給大家分享一些友盟在實踐中得到的一些經驗。

友盟實踐之數據采集

首先是從數據采集來說起，數據采集部分面臨了很大的挑戰。

第一是大流量，友盟服務大量開發者，接受數以億計的設備發來日志，每天要處理80億次對話，數據量非常大。
高并發也是移動互聯網的特點。比如：用戶在早上上班路上，中午吃飯后以及晚上睡覺前，是使用的高峰期，這個時候會對我們造成非常高的并發請求。
還有擴展性，因為移動互聯網是在高速發展的，最開始我們一臺機器就可以抗住，隨著發展我們現在可能需要10臺，20臺，如果系統沒有橫向擴展能力將會是很痛苦的事情。

友盟的數據平臺經歷了一個發展的過程。在2010年剛開始的時候，因為快速上線的要求，我們是基于RoR開發的，在后臺通過Resque進行一些離線的處理。這個架構，隨著互聯網的爆發，面臨巨大的數據壓力，很快就不能適用了。

接下來，我們就切換到基于Finagle Server的日志服務器。這個Finagle Server是Twitter開源出來的一個異步服務器框架，很適合移動互聯網的訪問特點：高并發，小數據量。切換到Finagle Server之后，我們單臺服務器的處理能力得到了極大的提升。同時日志收集服務的無狀態特性可以支持橫向擴展，所以當我們面臨非常高壓力的時候可以簡單地通過增加臨時服務器來解決。

友盟實踐之數據清洗

大數據的特點之一是數據多樣化，如果不進行清洗會對后面的計算產生困擾。在數據清洗方面，我們花了很多精力，并踩了很多的坑。

1、首先說“唯一標識”。

做數據分析，第一件事情就是要拿到“唯一標識”。安卓系統里作為唯一標識的，常用的是IMEI，MAC， AndroidID。首先因為安卓碎片化問題，通過API在采集這些數據的時候，常常會有采集不到的情況。

還有其他一些異常的情況，比如有很多山寨機沒有合法的 IMEI，所以會很多機器共用一個 IMEI，導致IMEI重復；有些 ROM 刷機后會更改 MAC 地址，導致 MAC 重復；大部分電視盒子本身就沒有IMEI。這種情況下我們單純用 IMEI 或者 MAC ，或者安卓 ID ，來進行標識的的話，就會出現問題。

友盟采用的辦法是由多帶帶的服務來統一計算。后臺會有離線任務來進行計算，發現重復率很高的標識符，加入到黑名單里。在計算的時候，直接跳過黑名單里的標識，換用另一種算法進行計算。

2、在“數據標準化”時，也遇到了很多的坑

比如：“設備型號”，并不是直接采集 model 這個字段就可以解決的。拿小米 3 舉例，這個手機會有很多版本，不同的批次 model 字段不一樣。對于這種情況，如果我們不進行統一標準化，我們算出來的結果，肯定有問題。

還會出現多機一型的情況，例如 m1，大家都知道是這是一款2011年出現的熱門手機，很有意思的是時過三年之后活躍設備數量發生了再次突增。經過調查發現，原來是他的一個對手廠家，在2014年底生產了一款暢銷的產品，model字段也叫m1。因此，我們就需要把設備型號，通過專門手段來和產品名稱對應上，統一標準化。

另外在數據標準化過程中，還會遇到“地域識別”的問題。地域識別是用 IP 地址來識別的。因為中國 IP 地址管理并不是非常規范，所以經常會出現，上一秒鐘大家還在北京，下一秒就到深圳的情況。對于解決這樣的問題，我們是選用設備一天中最常出現的 IP 地址作為當天的地域標識。

還有“時間識別”，也是很大的問題。最開始我們采用的都是客戶端時間。但是客戶時間有很大的隨意性，用戶的一個錯誤設置，就會導致時間不一致；另外一些山寨機會有 Bug，機器重啟之后，時間直接就變成1970年1月1號了；還有一種可能，產生數據的時候沒有網絡連接，當他重新聯網了，日志才會匯報到平臺，這樣的話數據就會產生延遲。

為了解決這些時間不一致的問題，我們統一使用服務器端時間。但是這樣又帶來了新的問題：統計時間和真實時間的差異，但是這個差異值是從小時間窗口（例如一個小時，或一天）觀察出來的，從大的時間窗口來看是正確的。

3、“數據格式的歸一化”也很重要。

因為我們 SDK，經過很多版的進化，上報上來的日志會有多種格式。早期的時候我們是采用 Json 格式，后期的話我們使用Thrift的格式。在數據平臺處理的時候，兩種格式切換很麻煩，所在我們在處理之前，我們把它統一成 Protobuf ，來進行后期計算。

友盟實踐之數據計算

在數據計算的時候，根據不同業務對于時延的容忍程度的高低，分為實時計算，離線計算和準實時計算。

實時計算，面臨的挑戰之一是時效性。因為實時計算是對延時非常敏感的，毫秒級的水平。如果說，你把不合適的計算，比如一些很耗cpu的計算放進來，會直接導致實時計算的延遲。所以在架構時，需要考量，把哪些放到實時部分合適，哪些不適合。另外實時計算往往會在寫數據庫時會產生IO延遲，需要對實時數據庫專門進行專門優化。我們實時計算部分選用了MongoDB存儲數據，同時不斷優化MongoDB的寫請求來解決這個問題。

另外一個挑戰是突發流量。用戶使用 App 的頻率并不均勻，早上、中午、晚上會有很高的使用率，尤其是晚上10:00-12:00這個時間段會對我們系統帶來非常大的壓力，得益于之前的架構設計，在達到一定的閾值之后，會觸發報警，運維的同學會進行臨時擴容來應對這些突發流量。
因為實時計算通常是增量計算，因此會產生誤差積累的問題。Lambda架構決定了實時和離線是兩套獨立的計算系統，所以必然會出現誤差。如果你長時間使用實時計算的結果，這個誤差會越來越大。我們現在解決的辦法是在實時處理時，不要給他太大的時間窗口，比如說最多不要超過一天，超過一天之后，就要開始清理，離線部分的計算每天計算一次，保證在這個時候離線部分的數據計算完成，這樣我們就可以用離線的數據來覆蓋實時數據，從而消除這個數據誤差。

離線計算，也會面臨一些問題。最常遇到的麻煩是數據傾斜問題。數據傾斜這個事情，幾乎是天然存在的，比如說一些大App的數據量，和小的App的數據量存在著巨大的差距，常常會在離線計算的時候產生長尾現象，并行的MR作業中總是有一兩個任務拖后腿，甚至超出單機計算能力。

產生數據傾斜的原因有很多種，針對不同的原因，有不同的解決辦法。最常見的原因是因為粒度劃分太粗導致的，比如說我們計算的時候，如果以App ID來進行分區，很容易導致數據傾斜。針對這種情況，友盟的解決辦法的是進行更細一步的劃分，比如說我們通過App ID 加設備ID進行分區，然后再將結果聚合起來，這樣就可以減少數據傾斜的發生。

第二個問題是數據壓縮的問題。離線計算的時候，往往輸入輸出都會很大，因此我們要注意時時刻刻進行壓縮，用消耗cpu時間來換取存儲空間的節省。這樣做能夠節省數據傳輸中的IO延遲，反而能夠降低整個任務的完成時間。

接下來會面臨資源調度的困難，因為我們各種任務優先級是不一樣的，比如一些關鍵的指標，要在特定時間算出來，有些任務則是早幾個小時都可以。Hadoop自帶的調度器無論是公平調度還是能力調度器都不能實現我們的需求, 我們是通過修改hadoop的調度器代碼來實現的。

另外我們還有一類任務對時延比較敏感，但是又不適合放到實時計算中的。這類任務我們稱之為準實時任務。例如報表的下載服務，因為是IO密集型任務，放入實時不太合適，但是它又對時間比較敏感，可能用戶等三五分鐘還是可以接受的，但是等一兩個小時就很難接受了。對于這些準實時任務我們之前采用的是通過預留一定資源來運行MR來實現的。現在用spark Streaming 專門來做這些事情。

在進行準實時計算時，里面也有一個資源占用的問題，在預留的過程中，會導致你的資源占用率過低，如何平衡是個問題；第二點很多實時計算的任務，往往也采用了增量計算模式，需要解決增量計算的誤差累計問題，我們通過一定時間的全量計算來彌補這個缺陷。

友盟實踐之數據存儲

數據存儲，根據我們之前的計算模式，我們也分為在線存儲和離線存儲兩部分。在實時部分的計算結果我們主要存在 MongoDB 里面，必須對寫IO進行優化。離線數據計算結果一般存儲在 HBase 里。但是HBase缺少二級索引。我們引入了Elastic Search，來幫助HBaes進行索引相關的工作。

在做數據服務的時候通過數據緩存能夠解決數據冷熱的問題。友盟數據緩存用的是 Redis，同時使用了 TwemProxy 來作負載均衡。友盟在數據緩存這方面的經驗就是需要預加數據，比如：每天凌晨計算完數據之后，在用戶真正訪問之前，需要把部分計算結果預先加載上去，這樣等到用戶訪問的時候，就已經在內存里了。

友盟實踐之數據增值

整個大數據的系統，價值最大的部分，就在于數據增值，對于友盟來說，目前數據增值主要分兩個大的方向。

1、首先是內部數據打通。

友盟實現了個性化“微”推送，基于用戶事件，結合用戶畫像、以及和阿里百川合作提供更多的維度信息，來為開發者提供更精準的推送。
比如：對一個汽車電商類的 App，可以圈定一部分有車的用戶來推送汽車配件相關信息；然后圈定一部分無車用戶來推送售車相關信息。

2、友盟在數據挖掘方面做了很多工作。

友盟針對現有的設備，進行了用戶畫像相關計算，通過用戶畫像能夠了解用戶的屬性和興趣，方便后續的數據橫向打通。同時我們提供了設備評級這個產品。這主要是針對一些作弊行為來設計的。比如說一些App，進行推廣的時候會發現，有些渠道的推廣數據會有不真實的情況。

為此通過數據平臺的統計算法和機器學習算法，把我們現有的所有設備，進行評級，哪些是垃圾設備，哪些是真實設備，能夠很好的識別出來。這樣一來，如果開發者有相關需求，我們可以提供設備評級相關指標，來幫助開發者測評這些推廣渠道，到底哪些可信，哪些不可信。

圍繞著這些渠道刷量的問題，我們還推出了健康指數。設備評級產品，只是從設備和渠道的角度，來觀察作弊的問題；而健康指數是從一個App整體的角度來觀察作弊情況。