基于開源生態(tài)系統(tǒng)的大數(shù)據(jù)平臺研究

雷 軍 葉航軍 武澤勝 張 鵬 謝 龍 何炎祥

1(武漢大學(xué)計算機學(xué)院 武漢 430072)2(小米科技有限責(zé)任公司 北京 100085)3 (軟件工程國家重點實驗室(武漢大學(xué)) 武漢 430072)(leijun@xiaomi.com)

基于開源生態(tài)系統(tǒng)的大數(shù)據(jù)平臺研究

雷 軍1,2葉航軍2武澤勝2張 鵬2謝 龍2何炎祥1,3

1(武漢大學(xué)計算機學(xué)院 武漢 430072)2(小米科技有限責(zé)任公司 北京 100085)3(軟件工程國家重點實驗室(武漢大學(xué)) 武漢 430072)(leijun@xiaomi.com)

大規(guī)模數(shù)據(jù)的收集和處理是近年的研究熱點，業(yè)界已經(jīng)提出了若干平臺級的設(shè)計方案，大量使用了開源軟件作為數(shù)據(jù)收集和處理組件.然而，要真正滿足企業(yè)應(yīng)用中海量數(shù)據(jù)存儲、多樣化業(yè)務(wù)處理、跨業(yè)務(wù)分析、跨環(huán)境部署等復(fù)雜需求，尚需設(shè)計具有完整性、通用性、支持整個數(shù)據(jù)生命周期管理的大數(shù)據(jù)平臺，并且對開源軟件進行大量的功能開發(fā)、定制和改進.從小米公司的行業(yè)應(yīng)用和實踐出發(fā)，在深入研究現(xiàn)有平臺的基礎(chǔ)上，提出了一種新的基于開源生態(tài)系統(tǒng)的大數(shù)據(jù)收集與處理平臺，在負(fù)載均衡、故障恢復(fù)、數(shù)據(jù)壓縮、多維調(diào)度等方面進行了大量優(yōu)化，同時發(fā)現(xiàn)并解決了現(xiàn)有開源軟件在數(shù)據(jù)收集、存儲、處理以及軟件一致性、可用性和效率等方面的缺陷.該平臺已經(jīng)在小米公司成功部署，為小米公司各個業(yè)務(wù)線的數(shù)據(jù)收集和處理提供支撐服務(wù).

Hadoop；開源生態(tài)系統(tǒng)；大數(shù)據(jù);數(shù)據(jù)中心；網(wǎng)絡(luò)虛擬化

大規(guī)模數(shù)據(jù)的收集和處理是近年來業(yè)界和學(xué)術(shù)界的熱點，被稱為“大數(shù)據(jù)”問題.“大數(shù)據(jù)”問題存在多種定義，現(xiàn)在普遍被接受的是IBM的3V定義[1]，即數(shù)量(volume)、種類(variety)和速度(velocity)，也就是數(shù)量巨大、種類豐富、快速生成并需要快速處理的數(shù)據(jù).大規(guī)模數(shù)據(jù)的收集和處理有許多實際的應(yīng)用.對互聯(lián)網(wǎng)企業(yè)而言，用戶在使用其產(chǎn)品的過程中會產(chǎn)生大量的業(yè)務(wù)數(shù)據(jù)，比如使用日志、交易日志和關(guān)系鏈等.對這些數(shù)據(jù)的分析和處理，可以深刻了解用戶的需求.每一次用戶對產(chǎn)品的使用都反映了用戶的需求和對產(chǎn)品的反饋.對這些數(shù)據(jù)的分析和挖掘可以幫助公司改進自身產(chǎn)品，提升用戶體驗，為用戶創(chuàng)造更大的價值.因此，公司通常有強烈的需求來分析和處理上述數(shù)據(jù).

以小米科技有限責(zé)任公司(以下簡稱：小米公司)為例，公司業(yè)務(wù)數(shù)據(jù)的收集、分析和處理是一個典型的大數(shù)據(jù)問題：PB量級的數(shù)據(jù)總量、多種數(shù)據(jù)格式(如逗號分隔值(comma separated value, CSV)、Thrift消息[2]、文本文件、關(guān)系數(shù)據(jù)庫等)、上百個數(shù)據(jù)來源、每日TB量級的數(shù)據(jù)增量和小時級別的處理速度要求等.大數(shù)據(jù)問題的解決，需要一套行之有效的技術(shù)架構(gòu)，一般是分層次的堆棧式技術(shù)架構(gòu).對此，EMC將大數(shù)據(jù)技術(shù)架構(gòu)分成了4層：基礎(chǔ)層、管理層、分析層和應(yīng)用層.小米公司的內(nèi)部業(yè)務(wù)在基礎(chǔ)層和管理層上沿襲了該框架，但在數(shù)據(jù)的應(yīng)用和分析上有所不同，以適應(yīng)公司自身的業(yè)務(wù)特點.

1) 數(shù)據(jù)存量和增量大.PB級別的數(shù)據(jù)總量和TB級別的數(shù)據(jù)日增量，對數(shù)據(jù)存儲和傳輸?shù)某杀九c效率提出很高的要求.

2) 業(yè)務(wù)線多、數(shù)據(jù)來源和格式多樣化.上百個業(yè)務(wù)項目和數(shù)據(jù)來源，多種異構(gòu)的數(shù)據(jù)格式，要求大數(shù)據(jù)平臺有足夠的靈活性和可擴展性.

3) 跨業(yè)務(wù)數(shù)據(jù)分析和挖掘的需求大.聯(lián)合利用用戶在多個產(chǎn)品上的使用數(shù)據(jù)，才能更深刻了解用戶的需求，更好地改善用戶體驗.

4) 業(yè)務(wù)部署和大數(shù)據(jù)平臺部署的情況比較復(fù)雜.多機房部署、異構(gòu)的機房環(huán)境、要求集群和平臺的部署、監(jiān)控和報警等要足夠高效.

本文從小米公司的應(yīng)用和實踐出發(fā)，在不失通用性的前提下，提出了一個基于開源生態(tài)系統(tǒng)[3]的統(tǒng)一的大數(shù)據(jù)收集和處理的基礎(chǔ)平臺.本文的主要貢獻是將開源軟件的組件與自主研發(fā)的軟件組成一個完整的大數(shù)據(jù)平臺，并通過一系列的技術(shù)創(chuàng)新和改進，使其能夠勝任真實場景下大數(shù)據(jù)對系統(tǒng)功能、性能、一致性和可用性等各方面的需求.本文首先介紹相關(guān)研究和實踐工作，然后分別描述平臺的總體架構(gòu)組成以及所做的改進和創(chuàng)新，最后展望未來的發(fā)展路線和計劃.

1 相關(guān)工作

關(guān)于大數(shù)據(jù)平臺，業(yè)界較為有代表性的工作是Facebook的實時數(shù)據(jù)收集和分析平臺[3-4].該平臺的目標(biāo)是解決大規(guī)模(scalability)和低延遲(latency)的問題，它既使用了Scribe[5]，HDFS(Hadoop distributed file system，是Hadoop項目的一個核心子項目[6])，MapReduce[7]，Hive[8]，HBase[9]等開源系統(tǒng)，也自行開發(fā)了Calligraphus，PTail，Puma等私有系統(tǒng).該平臺的側(cè)重點是數(shù)據(jù)的收集和匯聚，即實時的分類統(tǒng)計，而非通用的數(shù)據(jù)計算和分析服務(wù).這個平臺最終能夠在9 GBps的寫入速度下把延時控制在10 s之內(nèi).

學(xué)術(shù)界關(guān)于大數(shù)據(jù)平臺也有大量的研究和實踐，大致可以分為基于應(yīng)用、基于模型以及基于平臺3類.基于應(yīng)用的研究工作主要從Web日志挖掘這個應(yīng)用出發(fā)，考慮如何在Hadoop等開源的生態(tài)系統(tǒng)上構(gòu)建分布式、可存儲和挖掘大規(guī)模日志數(shù)據(jù)的平臺.主要的工作在于討論和驗證分布式集群對于提高Web日志挖掘效率的可行性，并提出了相應(yīng)的解決方案[10-12].基于模型的工作重點是討論了更為通用的海量數(shù)據(jù)處理和計算模型，包括計算模型本身、網(wǎng)絡(luò)模型和優(yōu)化、編程模型等關(guān)鍵問題，也討論了通用模型在具體應(yīng)用中的實際問題和效果，比如數(shù)據(jù)清洗、容錯等[13-15].此外，基于平臺的工作更多是從平臺自身的角度，比如數(shù)據(jù)管理、資源調(diào)度與虛擬化，并把整個系統(tǒng)分成多個層次[16-18].例如把系統(tǒng)分為數(shù)據(jù)庫訪問層、數(shù)據(jù)處理層和業(yè)務(wù)應(yīng)用層[16]；將系統(tǒng)分為算法層、任務(wù)層和用戶層[18].

目前已有的工作主要集中研究了大數(shù)據(jù)平臺中一些重要組件的設(shè)計和實現(xiàn).由于小米公司的業(yè)務(wù)具有數(shù)據(jù)量大、業(yè)務(wù)需求多樣化、跨業(yè)務(wù)分析的需求大、部署環(huán)境復(fù)雜等特點，需要一個能管理海量數(shù)據(jù)整個生命周期的、完整的、通用的大數(shù)據(jù)平臺.此外，還需解決現(xiàn)有的系統(tǒng)在數(shù)據(jù)收集、存儲和處理、一致性、可用性和效率等關(guān)鍵問題上存在的缺陷.然而，現(xiàn)有開源軟件的組合方案在數(shù)據(jù)存儲、壓縮、傳輸?shù)刃阅苌铣３o法滿足大型互聯(lián)網(wǎng)企業(yè)的海量業(yè)務(wù)處理需求；另一方面，現(xiàn)有方案也無法支持多樣化業(yè)務(wù)的分析和挖掘需求.此外，分布式部署環(huán)境下的可靠性尚需提升，存儲、帶寬、維護成本也需要進行優(yōu)化.

本文從通用平臺設(shè)計的角度出發(fā)，主要解決下列問題：大規(guī)模數(shù)據(jù)的實時收集和存儲、計算資源與作業(yè)的管理與調(diào)度、集群管理(部署、監(jiān)控和報警).同時，在功能、一致性、可用性和效率等方面做了重大改進和提高.

2 總體架構(gòu)

一個完整通用的大數(shù)據(jù)平臺，至少要涵蓋數(shù)據(jù)的收集、存儲、計算和管理等方面.本平臺選用了部分開源軟件作為系統(tǒng)的主要組件，包括ZooKeeper[19]，Hadoop(HDFSMapReduceYARN)，HBase，Hive， Scribe等.這些開源軟件相對成熟，生態(tài)系統(tǒng)已經(jīng)比較完備，可用于快速搭建大數(shù)據(jù)平臺.在此基礎(chǔ)上，本平臺增加了自主開發(fā)的Minos監(jiān)控系統(tǒng)，并基于對業(yè)務(wù)特性的深入分析調(diào)整和完善平臺的設(shè)計.圖1是平臺的整體架構(gòu)圖.出于完整性的考慮，該架構(gòu)圖還包含了該大數(shù)據(jù)平臺正在試驗支持的計算框架，包括Storm[20]，Spark[21]，Impala[22]等.

Fig. 1 Overall architecture of big-data platform圖1 大數(shù)據(jù)平臺整體架構(gòu)圖

3 數(shù)據(jù)收集系統(tǒng)

對于大部分應(yīng)用場景來說，業(yè)務(wù)數(shù)據(jù)的來源和格式經(jīng)常會有很多種，比如Apache或Nginx等Web Server的訪問日志、業(yè)務(wù)自定義的CSV格式文件以及用Protocol Buffer[23]或者Thrift消息編碼過后的消息.一個足夠通用和靈活的數(shù)據(jù)收集平臺，需要同時滿足不同業(yè)務(wù)的多樣化需求.

許多開源的數(shù)據(jù)收集系統(tǒng)，比如Facebook的Scribe[5]、LinkedIn的Kafka[24]、Cloudera的Flume[25]和Apache的Chukwa[26]，在業(yè)界都有廣泛的應(yīng)用.如果需要考慮到業(yè)務(wù)種類較多，數(shù)據(jù)格式和對數(shù)據(jù)的后續(xù)處理有多種方式，期望的數(shù)據(jù)收集系統(tǒng)需要滿足下面6個特點(優(yōu)先級由高到低)：

1) 高可用.數(shù)據(jù)不會因為單節(jié)點或者少數(shù)節(jié)點的故障丟失.

2) 靈活.能夠滿足多種業(yè)務(wù)不同的使用方式和后續(xù)處理需求.

3) 使用簡單.各業(yè)務(wù)接入系統(tǒng)的學(xué)習(xí)成本較低.

4) 易配置和維護.較低的運維成本.

5) 低外部依賴.較低的運維成本.

6) 架構(gòu)和實現(xiàn)簡單.多數(shù)開源系統(tǒng)需要一些改進來適配業(yè)務(wù)的要求.

綜合考慮，Scribe在這6個方面有一定優(yōu)勢，圖2是本文提出的基于Scribe的數(shù)據(jù)收集系統(tǒng)架構(gòu)圖.

Fig. 2 Architecture of data collection system圖2 數(shù)據(jù)收集系統(tǒng)架構(gòu)圖

3.1 數(shù)據(jù)傳輸?shù)膬?yōu)化與改進

在設(shè)計支持跨數(shù)據(jù)中心的分布式數(shù)據(jù)收集系統(tǒng)時，為了統(tǒng)計和數(shù)據(jù)處理的方便，經(jīng)常需要將所有的業(yè)務(wù)數(shù)據(jù)最終寫入到同一個Hadoop集群里(也會在同一個數(shù)據(jù)中心)，引起跨數(shù)據(jù)中心的數(shù)據(jù)傳輸.實踐發(fā)現(xiàn)，大量的日志數(shù)據(jù)占據(jù)跨數(shù)據(jù)中心帶寬的相當(dāng)比例，浪費了寶貴的帶寬資源.

本文提出了一種改進方法，可以在傳輸時對收集的數(shù)據(jù)進行壓縮.實踐證明這可以有效地減少數(shù)據(jù)傳輸量，很大地節(jié)約運營成本.

Scribe是通過Thrift的RPC接口對外提供服務(wù)，Thrift本身不提供傳輸數(shù)據(jù)壓縮的功能.Thrift本身也是一個分層設(shè)計的結(jié)構(gòu)，加上Scribe又是搭建在Thrift之上的應(yīng)用，所以有多個地方可以選擇來實現(xiàn)壓縮，比如Thrift Protocol層、Thrift Transport層或者在Scribe本身.由于其他Thrift Server也可能有數(shù)據(jù)傳輸壓縮的需求，本文提出了一種通用的解決方案，在Thrift Transport層來實現(xiàn)Compressed的傳輸協(xié)議，使得各類Thrift Server都能與之兼容.

Thrift本身提供了良好的擴展性.Thrift Server缺省使用了內(nèi)置的TFramedTransport傳輸協(xié)議，這是一個直接基于系統(tǒng)底層傳輸協(xié)議(在Thrift Server里就是TCP協(xié)議)之上的簡單的非壓縮傳輸協(xié)議.同時Thrift Server在構(gòu)造的時候允許傳入一個TTransportFactory的傳輸層工廠類，通過傳輸層的串聯(lián)模式，可以在內(nèi)置傳輸協(xié)議的基礎(chǔ)上實現(xiàn)更復(fù)雜的協(xié)議.

Fig. 3 Default transport protocol and compressed transport protocol圖3 缺省傳輸協(xié)議與壓縮傳輸協(xié)議

本文提出了一種新壓縮傳輸協(xié)議TSnappy-Transport和它的工廠類TSnappyTransportFactory.圖3是原始的非壓縮的傳輸協(xié)議和本文提出的壓縮傳輸協(xié)議的對比.由于本文提出的協(xié)議使用了傳輸層的串聯(lián)模式，所以可以認(rèn)為在原始的傳輸協(xié)議基礎(chǔ)上，對它的有效載荷(payload)又進行了一次分塊壓縮與編碼.

本文提出的壓縮傳輸協(xié)議使用了Snappy壓縮算法，它是Google提出并開源的一個壓縮算法和代碼庫[27].和其他常用的壓縮算法相比，它的最大特點是在壓縮率可接受的情況下，壓縮和解壓縮的速度非?？?例如與zlib的快速模式相比，對于大部分輸入Snappy能夠快10倍以上，但其壓縮率會有20%～50%的損失.所以該算法特別適用于在線傳輸數(shù)據(jù)的壓縮，不會給CPU造成嚴(yán)重負(fù)擔(dān)或明顯增加延遲.

根據(jù)Google的官方數(shù)據(jù)，使用64位Intel Core i7 CPU，單核模式下Snappy的壓縮速度超過250 MBps，解壓速度超過500 MBps.線上服務(wù)器一般是8～24核的配置，所以它引起的CPU開銷基本可以忽略不計.

目前的實現(xiàn)僅支持了一種壓縮算法，所以本文提出的壓縮傳輸層協(xié)議直接命名為Snappy Transport.理論上該協(xié)議可以擴展支持任意的塊壓縮算法，以便于業(yè)務(wù)根據(jù)實際需求進行選擇，留給將來的工作做擴展.

表1是從3種典型的業(yè)務(wù)日志數(shù)據(jù)中分別抽取一段，分別用未壓縮和壓縮2種模式傳輸日志消耗的網(wǎng)絡(luò)帶寬以及壓縮率.

Table 1 Compression Ratio of Data Transportation

在真實業(yè)務(wù)場景下，壓縮傳輸只使用了原來30%左右的網(wǎng)絡(luò)帶寬，并且CPU沒有成為新的瓶頸，因此也不需要部署新的Scribe Server來分擔(dān)負(fù)載.該項改進明顯降低了日志數(shù)據(jù)在網(wǎng)絡(luò)傳輸上的成本.

3.2 負(fù)載均衡和故障處理的優(yōu)化與改進

數(shù)據(jù)收集系統(tǒng)很重要的一個要求是高可用性.Scribe在這方面有獨特設(shè)計，比如Buffer Store可以在下游的主通道不可用的時候，先把數(shù)據(jù)寫到本地文件(也可以配置為寫到其他Store中)，待下游主通道可用時再把本地緩存的數(shù)據(jù)發(fā)送過去.

在本文提出的數(shù)據(jù)收集系統(tǒng)中，需要有一套中心服務(wù)器負(fù)責(zé)接受所有業(yè)務(wù)的數(shù)據(jù)，再把數(shù)據(jù)寫入到統(tǒng)一的HDFS集群中.為了避免該服務(wù)器成為系統(tǒng)的故障點，需要用一主一備2個服務(wù)器來提高可用性，用Buffer Store配置成主服務(wù)器不可用時寫入備服務(wù)器.這在應(yīng)對服務(wù)器的偶然宕機或者運維操作時將起關(guān)鍵作用，顯著提升可用性.

然而，在這種配置下的單個服務(wù)器需要承擔(dān)系統(tǒng)的所有負(fù)載(主和備同時只有一個在提供服務(wù)).隨著業(yè)務(wù)數(shù)據(jù)流量的增加，在業(yè)務(wù)峰值時，流量經(jīng)常超過單個服務(wù)器的處理能力.如果主服務(wù)器因為超載變得不可用，所有數(shù)據(jù)又都會寫到備服務(wù)器，由于這些服務(wù)器的配置相同，備服務(wù)器也常常超載，導(dǎo)致整個系統(tǒng)的不可用或者抖動.實際上并不需要關(guān)心具體是哪個Scribe服務(wù)器把數(shù)據(jù)寫入到HDFS，所有服務(wù)器的角色是對等的，所以需要一個完備的負(fù)載均衡方案.Scribe有一種Bucket Store的配置，具有負(fù)載均衡的能力，但對Scribe服務(wù)器的故障處理(failover)支持差，單個服務(wù)器故障也會導(dǎo)致整個系統(tǒng)不可用.本文對此提出了4點改進以提高可用性:

1) 跟蹤所有服務(wù)器的狀態(tài)，未能成功應(yīng)答的服務(wù)器會被標(biāo)志成“不可用”.

2) 只有處于“可用”狀態(tài)的服務(wù)器才會成為日志數(shù)據(jù)下發(fā)的候選.

3) 定義了一種“round_robin”的Bucket新類型，在所有“可用”服務(wù)器中循環(huán)選擇候選下發(fā)數(shù)據(jù)，直到有一個服務(wù)器成功應(yīng)答(即發(fā)送成功).

下面通過模擬實驗來比較改進前后日志收集系統(tǒng)的總體可用性.假設(shè)單個Scribe服務(wù)器的可用性為p，總共有n臺Scribe服務(wù)器，將n臺Scribe服務(wù)器配成n個bucket.假設(shè)各個服務(wù)器的可用性是獨立的，可以推導(dǎo)出總體可用性為

(1)

在改進之后，同樣假設(shè)各個服務(wù)器的可用性是獨立的，但至少要有m個服務(wù)器可用總體系統(tǒng)才可用(考慮到服務(wù)器的處理能力)，可以推導(dǎo)出總體可用性為

(2)

表2比較了改進前后日志收集系統(tǒng)的總體可用性.假設(shè)單個Scribe服務(wù)器的可用性p=0.99，同時至少有一半的服務(wù)器可用總體系統(tǒng)才可用(m=n2).這個改進徹底解決了Scribe在負(fù)載均衡和故障處理上的缺陷.在業(yè)務(wù)中的實踐也表明進行上述改進后，可用性和系統(tǒng)的可擴展性有明顯提高，沒有再出現(xiàn)因為超載或者單機故障造成的系統(tǒng)不可用.

Table 2 Comparison of Log Collection System OverallAvailability BeforeAfter Improvement

表2 改進前后日志收集系統(tǒng)總體可用性的比較

(p=0.99,m=n2)

Table 2 Comparison of Log Collection System OverallAvailability BeforeAfter Improvement

nBeforeImprovementAfterImprovement20．98010．999940．960596010．9999960360．9414801494010．99999985239

4 數(shù)據(jù)存儲系統(tǒng)

數(shù)據(jù)規(guī)模較大的存儲會超出單機的存儲能力，需要一個分布式的存儲系統(tǒng).傳統(tǒng)的技術(shù)包括存儲區(qū)域網(wǎng)絡(luò)(storage area network, SAN)、網(wǎng)絡(luò)附加存儲(network attached storage, NAS)、網(wǎng)絡(luò)文件系統(tǒng)(network file system, NFS)等.這些存儲技術(shù)都需要高端或?qū)Ｓ么鎯υO(shè)備，成本通常較高.

近年來隨著低成本存儲設(shè)備的可靠性提高，軟件冗余和糾錯技術(shù)的發(fā)展，也逐漸出現(xiàn)了基于廉價和通用存儲設(shè)備的分布式文件系統(tǒng).尤其是Google發(fā)表了內(nèi)部設(shè)計和使用的分布式文件系統(tǒng)(Google file system, GFS)[28]，驗證了這種技術(shù)在提供類似可靠性的前提下，性價比和可擴展性有很大的提高.

此后出現(xiàn)了大量的開源實現(xiàn).其中HDFS是使用比較廣泛、也比較成熟的一種開源實現(xiàn).本文提出的大數(shù)據(jù)平臺也是以HDFS為核心的存儲系統(tǒng).

作為一個分布式存儲系統(tǒng)，最重要的衡量指標(biāo)是一致性(可靠性)、可用性和性能.尤其是一致性和可用性，往往是選擇一個分布式存儲系統(tǒng)時的關(guān)鍵因素.在部署和使用開源的HDFS版本時，我們發(fā)現(xiàn)HDFS在一致性和可用性上的一些嚴(yán)重缺陷.本文提出了相應(yīng)的改進和優(yōu)化方案并在業(yè)務(wù)系統(tǒng)中部署了改進后的版本.

4.1 一致性的優(yōu)化與改進

存儲系統(tǒng)由于各種原因(新特性、修復(fù)缺陷等)，會對軟件版本進行發(fā)布和升級.為了盡量避免對業(yè)務(wù)的影響和提高可用性，更好的實踐是在持續(xù)提供服務(wù)的情況下，對集群中的各個節(jié)點進行逐臺滾動升級.

德斯拜思機電控制技術(shù)（上海）有限公司是德國dSPACE于2008年在中國建立的分支機構(gòu)。20多年以來，德國dSPACE的高品質(zhì)現(xiàn)成軟件和硬件工具使工程師可以隨心所欲地進行設(shè)計和創(chuàng)新，并顯著減少了開發(fā)時間和成本。憑借廣泛的產(chǎn)品系列和高新技術(shù)，該公司成為汽車工業(yè)、航空航天領(lǐng)域和工業(yè)自動化領(lǐng)域最受歡迎的開發(fā)合作伙伴之一。

在實施過程中發(fā)現(xiàn)在這種升級方式下，HDFS上的文件很小概率下有損壞的情況.對于一個存儲系統(tǒng)而言，文件損壞是很嚴(yán)重的缺陷，所以也是本文必須要解決的問題.由于該現(xiàn)象是偶發(fā)的，深入分析后確認(rèn)是在HDFS寫數(shù)據(jù)的流水線中間節(jié)點宕機后恢復(fù)的過程中，由于HDFS本身邏輯的缺陷，導(dǎo)致Checksum文件多出一個Checksum，從而導(dǎo)致HDFS校驗Checksum失敗，進而認(rèn)為數(shù)據(jù)被損壞.這已經(jīng)相當(dāng)于出現(xiàn)了丟失數(shù)據(jù)的現(xiàn)象，之前已經(jīng)成功寫入的數(shù)據(jù)無法再正確讀出，從而破壞了一致性的約定.

在Hadoop2.0版本時我們已向社區(qū)匯報了該問題，并提交了補丁代碼[29].該問題被社區(qū)確認(rèn)為嚴(yán)重的數(shù)據(jù)損壞問題，并在Hadoop2.7版本中得到了解決.對此缺陷進行了修正之后，再未出現(xiàn)集群逐臺滾動升級時的文件損壞.

除了需要對存儲系統(tǒng)的軟件版本進行升級外，經(jīng)常也會有需求添加或移除一些存儲節(jié)點(DataNode).添加存儲節(jié)點的過程比較簡單，只需要在新的節(jié)點上配置好軟件環(huán)境并啟動相應(yīng)的服務(wù)即可，將來的數(shù)據(jù)寫入就會依據(jù)一定的概率和規(guī)則分配到新節(jié)點上.但移除舊節(jié)點會復(fù)雜一些，為了防止數(shù)據(jù)丟失或者可靠性下降，需要先將舊節(jié)點所服務(wù)的數(shù)據(jù)移到還將提供服務(wù)的節(jié)點之后才能下線.同樣的，需要存儲集群在整個移除過程仍能正常服務(wù).

HDFS提供了從集群優(yōu)雅地卸下存儲節(jié)點的機制(decommission).在集群遷移的過程中，需要同時卸下(decommission)多個節(jié)點.實施過程中發(fā)現(xiàn)，當(dāng)Decommission進行到最后的時候，有部分節(jié)點無法結(jié)束Decommission，強制把這些節(jié)點關(guān)閉服務(wù)發(fā)現(xiàn)會有數(shù)據(jù)丟失.經(jīng)過調(diào)查發(fā)現(xiàn)，在移除節(jié)點的過程中，如果某個數(shù)據(jù)塊的3個副本都在需要移除的節(jié)點上，而且這個數(shù)據(jù)塊在移除時正在被打開寫的話，這里HDFS自身的處理邏輯有缺陷，會導(dǎo)致這樣的數(shù)據(jù)塊無法被正常復(fù)制到能夠提供正常服務(wù)的節(jié)點上去.

針對該缺陷，本文調(diào)整了文件完成的判斷條件：只要活躍節(jié)點和待移除節(jié)點上的塊復(fù)本數(shù)滿足最小復(fù)本數(shù)，則正常結(jié)束文件.之后由Decommision流程將數(shù)據(jù)塊從待移除節(jié)點復(fù)制到活躍節(jié)點，完成全部數(shù)據(jù)塊復(fù)制后再移除節(jié)點，實現(xiàn)了無數(shù)據(jù)損失的節(jié)點退出.

下面通過模擬實驗計算在改進之前出現(xiàn)異常(移除節(jié)點時無法正常結(jié)束或者丟失數(shù)據(jù))的概率.假設(shè)集群有n個存儲節(jié)點，同時移除m個存儲節(jié)點，當(dāng)時有k個文件同時被寫入數(shù)據(jù).根據(jù)前面的分析，只要任何一個正在被寫入的文件的3個副本都在這m個存儲節(jié)點，就會出現(xiàn)異常.假設(shè)副本在數(shù)據(jù)節(jié)點上的分配是均勻分布且獨立的，可以推導(dǎo)出出現(xiàn)異常的概率為

(3)

Table 3 Probability of Abnormity for RepresentativeConfigurations

表3給出了5種典型配置下出現(xiàn)異常的概率.可以看出在對此缺陷進行了修正之前，出現(xiàn)異常導(dǎo)致移除節(jié)點時無法正常結(jié)束或者丟失數(shù)據(jù)的概率較大.對此缺陷進行了修正之后，再未出現(xiàn)集群移除節(jié)點時無法正常結(jié)束或者丟失數(shù)據(jù)的情況.

4.2 可用性的優(yōu)化與改進

當(dāng)前HDFS的實現(xiàn)中，在客戶端有一個數(shù)據(jù)節(jié)點(DataNode)的黑名單，在用戶使用客戶端操作HDFS的過程中，如果發(fā)現(xiàn)某個數(shù)據(jù)節(jié)點出現(xiàn)故障，都會被加入到這個黑名單，后續(xù)該客戶端就不再從該數(shù)據(jù)節(jié)點讀寫數(shù)據(jù).這樣是一種優(yōu)化，目的是避免從故障或者繁忙的節(jié)點讀寫數(shù)據(jù).

在集群規(guī)模較小時，由于集群上的計算任務(wù)繁重，高負(fù)載的情況時有發(fā)生，導(dǎo)致客戶端偶爾發(fā)生數(shù)據(jù)節(jié)點讀、寫超時的情況.這類數(shù)據(jù)節(jié)點將被加入到上述黑名單.在本文的數(shù)據(jù)收集系統(tǒng)中，中央Scribe Server寫HDFS的模式是：打開一個文件持續(xù)寫，直到達到一定的大小，或者到第2天再切換文件.在實際的生產(chǎn)環(huán)境中，有些業(yè)務(wù)數(shù)據(jù)量不大但持續(xù)會有，一天的日志總大小達不到切換文件的條件，因此，一整天都在持續(xù)地寫同一個文件.在這樣的情況下，當(dāng)所有的數(shù)據(jù)節(jié)點都進入到黑名單后，Scribe Server對HDFS就不能寫了.由于這個黑名單是文件流級別的，所以后續(xù)除非重新創(chuàng)建文件流，否則該文件流涉及到數(shù)據(jù)節(jié)點的操作都會失敗.這時已經(jīng)寫入的數(shù)據(jù)不會丟失，而且能夠正確讀出，但從Scribe Server的角度，HDFS集群已經(jīng)處于不可用的狀態(tài).

下面通過模擬實驗來計算在改進之前HDFS集群出現(xiàn)不可用的概率.假設(shè)集群有n個存儲節(jié)點，每個存儲節(jié)點在這個時間周期內(nèi)(這里是1 d)出現(xiàn)不可用(主要是讀寫超時)的概率為p，這個時間周期內(nèi)有k個文件被寫入數(shù)據(jù)且未出現(xiàn)文件切換.假設(shè)副本在數(shù)據(jù)節(jié)點上的分配是均勻分布且獨立的，存儲節(jié)點出現(xiàn)不可用是獨立事件，可以推導(dǎo)出HDFS集群出現(xiàn)不可用的概率為

(4)

表4給出了6種典型配置下出現(xiàn)不可用(某個文件無法寫入)的概率.

在優(yōu)化和改進之前，HDFS集群有較高的概率出現(xiàn)某個文件不可寫入.在本平臺中，存儲與計算共享同一個集群，而且集群上的計算任務(wù)大，單個機器在1d的時間周期里，出現(xiàn)(對某個客戶端至少一次)讀寫超時的概率非常高.另外計算任務(wù)是批處理提交的，機器出現(xiàn)讀寫超時并不是獨立的，所以會經(jīng)常遇到某個文件不可寫入的情況.

Table 4 Probability of Unavailability for RepresentativeConfigurations

對此本文做了優(yōu)化和改進，對于進入黑名單的數(shù)據(jù)節(jié)點，當(dāng)它進入黑名單超過一定的時間，給與它一定的機會讓其復(fù)活.從上線后的效果來看，對可用性有很明顯的提高，再未出現(xiàn)由于數(shù)據(jù)節(jié)點負(fù)載高造成的偶爾超時，導(dǎo)致某個文件不可寫入的情況.

5 計算系統(tǒng)

和分布式的數(shù)據(jù)存儲系統(tǒng)相類似，對規(guī)模較大的數(shù)據(jù)進行處理和計算，往往也會超出單機的處理能力，需要一個并行計算的系統(tǒng)和框架，傳統(tǒng)的技術(shù)包括MPI和分布式數(shù)據(jù)庫等.

Google近些年陸續(xù)發(fā)表了內(nèi)部設(shè)計和使用的計算框架，包括MapReduce，Sawzall[30]，Dremel[31]等，為大規(guī)模數(shù)據(jù)的計算框架帶來了一些新思路.其中MapReduce是把所有的并行計算都分解為Map，Shuffle和Reduce這3個階段進行并行化，能夠滿足一大類并行計算的需求；而Dremel則是用SQL語句來表示計算任務(wù)，由后臺的計算系統(tǒng)把SQL語句翻譯成執(zhí)行計劃，在多個節(jié)點上并行執(zhí)行.這2種框架非常適合大規(guī)劃數(shù)據(jù)的批次處理.

在開源生態(tài)系統(tǒng)里，Hadoop的MapReduce(也是Hadoop項目的一個核心子項目)和Hive是對應(yīng)的2個實現(xiàn)，也是目前使用廣泛、成熟度較高的實現(xiàn).本文提出的大數(shù)據(jù)平臺，也是以開源的MapReduce和Hive為核心的計算系統(tǒng).

在具體的MapReduce版本方面本文選用了最新的Hadoop MapReduce 2.0，該版本引入了通用的資源調(diào)度系統(tǒng)YARN，整體架構(gòu)也代表了下一代計算和資源管理的發(fā)展方向，也得到了業(yè)界的廣泛認(rèn)可和支持.在2.0的架構(gòu)中，資源調(diào)度和作業(yè)調(diào)度邏輯分離，有效地減輕了中央節(jié)點的壓力，以提供更好的集群可擴展性.各個MapReduce作業(yè)之間是獨立的流程，由各自的Job Master進行管理，單個作業(yè)的失敗不會影響到其他作業(yè)，因此作業(yè)的容錯方面較1.0的架構(gòu)也有了大幅改進.另外相比先前架構(gòu)中以槽位(slot)作為單一調(diào)度維度，新架構(gòu)中引入了內(nèi)存、CPU等多個調(diào)度維度，用戶可以更準(zhǔn)確地對任務(wù)所需要的資源進行描述，有利于集群資源的有效利用.此外，2.0架構(gòu)中的通用資源系統(tǒng)還支持在其上運行多種非MapReduce的作業(yè)，這也為不同業(yè)務(wù)的集群復(fù)用提供了可能.

5.1 計算資源的配額管理

在本平臺的Hadoop應(yīng)用中，離線集群存儲了多種業(yè)務(wù)的數(shù)據(jù)，各業(yè)務(wù)通常都有各自的計算處理需求.除了HDFS 存儲配額管理之外，還需要為各業(yè)務(wù)的計算需求合理地分配計算資源.

Hadoop的YARN延續(xù)了之前MapReduce的調(diào)度器的模型，包括先入先出調(diào)度器(FifoScheduler)、容量調(diào)度器(CapacityScheduler)以及公平調(diào)度器(FairScheduler).先入先出調(diào)度器是系統(tǒng)的默認(rèn)調(diào)度器，它不考慮作業(yè)間的優(yōu)先級差異，簡單地按先到先服務(wù)的策略進行作業(yè)調(diào)度，在前面的作業(yè)沒有執(zhí)行完前，后續(xù)的作業(yè)只能排隊等待，因此它并不適合本文所討論的企業(yè)級需求場景；容量調(diào)度器和公平調(diào)度器在演化的過程中相互取長補短，功能特性具有一定的相似性，它們相比默認(rèn)的調(diào)度器支持作業(yè)的優(yōu)先級設(shè)置，支持多級調(diào)度隊列的配置，支持作業(yè)搶占等，適用于企業(yè)級集群的資源分配場景.考慮到公平調(diào)度器還在開發(fā)和完善階段，本文選用了更成熟的容量調(diào)度器作為資源配額管理的方案.

在實踐中，面對不同業(yè)務(wù)的計算需求，本平臺為各主要業(yè)務(wù)建立作業(yè)隊列，為每個隊列配置一定的計算資源底限以保證基本運算需求，同時為每個隊列設(shè)置允許在集群空閑時最多使用的資源量，以提高集群整體的利用率.考慮到業(yè)務(wù)的層次化結(jié)構(gòu)，本平臺還在一級作業(yè)隊列下建立二級隊列，以滿足一個業(yè)務(wù)內(nèi)部的細(xì)分計算需求.通過隊列的合理配額配置，在對各業(yè)務(wù)的資源需求進行隔離的同時，也能夠充分復(fù)用集群，最大化集群的資源利用率.

5.2 多維度資源調(diào)度

在Hadoop 1.0中，計算資源使用槽位作為表示方式.一個計算節(jié)點上的CPU、內(nèi)存等資源被等分為若干個槽位，每個任務(wù)則描述需求多少個槽位的資源.這種方式將多維度的資源抽象為一種“資源”，簡化了資源調(diào)度問題，但這種方式也有很多不足：槽位是預(yù)先靜態(tài)劃分的，無法最佳地適應(yīng)動態(tài)變化的作業(yè)，通常導(dǎo)致由于劃分粒度過大而造成資源的浪費；其次，單一維度的資源描述不利于對CPU或內(nèi)存需求多樣化的任務(wù)共享資源，降低了集群的資源利用率；另外，以槽位作為資源描述單位也不方便對任務(wù)進行使用資源的隔離.

針對基于槽位調(diào)度的不足，Hadoop 2.0的YARN引入了多維度的資源調(diào)度，目前支持CPU和內(nèi)存2個維度.例如，在新框架下，一個偏內(nèi)存型的任務(wù)可以描述它需要4 GB的內(nèi)存和1個CPU核，而偏CPU型的任務(wù)可以描述它需要1 GB內(nèi)存和4個CPU核，這樣的2個任務(wù)在不同維度上的需求互補性，可以最大化地發(fā)揮計算節(jié)點的資源利用率.除了充分提高資源利用率的同時，多維度的資源調(diào)度也有利于控制一個節(jié)點的并發(fā)任務(wù)，避免讓節(jié)點負(fù)載過高.假設(shè)在集群中節(jié)點的內(nèi)存較大(如64 GB)，而CPU核數(shù)較少(如8核)，在只有內(nèi)存一個維度調(diào)度的情況下，要求1 GB內(nèi)存的任務(wù)會在一個節(jié)點上運行幾十個，任務(wù)彼此間會形成對CPU資源的強烈競爭，導(dǎo)致機器負(fù)載高，作業(yè)執(zhí)行速率也大幅下降.引入CPU維度后，任務(wù)默認(rèn)指定需求一個CPU核，調(diào)度時會因在這一維度達到上限而不再下發(fā)任務(wù)，從而控制機器的負(fù)載，保證作業(yè)的計算性能.

多維度調(diào)度的引入大大優(yōu)化了資源的描述和資源調(diào)度功能，但由于它是Hadoop 2.0中較新的特性，所以也有一些潛在的問題.例如在使用過程中發(fā)現(xiàn)它在調(diào)度時計算下發(fā)任務(wù)量時存在缺陷，可能會而導(dǎo)致MapReduce作業(yè)的調(diào)度死鎖.針對這一較嚴(yán)重的問題，本文對容量調(diào)度器進行了修改，在下發(fā)時綜合多維度資源計算下發(fā)任務(wù)量，從而避免了調(diào)度死鎖的發(fā)生.

5.3 容量調(diào)度器的負(fù)載均衡

容量調(diào)度器的功能滿足了本平臺的大部分需求，但它也存在不完善的地方.在實踐中，調(diào)度器會在計算節(jié)點心跳匯報時，盡可能多地下發(fā)任務(wù).這一策略不利于計算任務(wù)在集群中的均勻分布：在集群整體空閑時，任務(wù)集中分布在少量的節(jié)點上，并沒有充分利用集群中節(jié)點的并發(fā)計算能力.針對這一問題，本文修改了調(diào)度下發(fā)策略，限制單節(jié)點單次下發(fā)的任務(wù)上限.修改后雖然會降低平均下發(fā)的速率，但由于任務(wù)在集群中的分布更新均勻，有效地利用了節(jié)點間的并發(fā)，因此整體上縮短了作業(yè)級的執(zhí)行時間：在集群空閑時單作業(yè)執(zhí)行時間能縮短30%～50%.另外引入單次下發(fā)的上限，在一定程度上也避免了內(nèi)存或CPU需求密集性的任務(wù)集中分布在單個節(jié)點，有利于使一個節(jié)點上的任務(wù)需求多樣化，提高單節(jié)點上可運行的任務(wù)數(shù)和節(jié)點資源的利用率.

5.4 MapReduce開發(fā)流程優(yōu)化

在離線處理集群的運營過程中，除了積累Hadoop系統(tǒng)的應(yīng)用和改進經(jīng)驗之外，對于優(yōu)化MapReduce開發(fā)流程本文也進行了探索和嘗試.分布式環(huán)境中，當(dāng)程序出現(xiàn)問題時，快速準(zhǔn)確地定位問題是一個巨大挑戰(zhàn).通常情況下，MapReduce程序的開發(fā)者在編寫完程序后會在集群上直接運行測試，當(dāng)出現(xiàn)異常時，很多時候需要查看作業(yè)日志，甚至到遠(yuǎn)程計算節(jié)點上分析問題.這種方式的問題定位成本非常高，既耗費了開發(fā)者的大量時間，也浪費了寶貴的集群計算資源.在協(xié)助用戶定位問題的過程中發(fā)現(xiàn)，很多問題并不需要在集群上運行作業(yè)才能暴露出來，通過單元測試或本地模式運行就可以有效地排查.因此本文提出一個優(yōu)化的開發(fā)流程如下：

1) 開發(fā)程序時，利用MR Unit測試框架為Mapper和Reducer等編寫單元測試.通過單元測試覆蓋主要場景，保證程序的基本正確性.

2) 取部分真實輸入數(shù)據(jù)，利用MapReduce的本地模式運行作業(yè)，排查真實數(shù)據(jù)中的邊界情況.如果遇到錯誤，則可以利用Eclipse等集成開發(fā)環(huán)境單機調(diào)試，分析定位問題.之后可以把新的場景補充到單元測試之中.

3) 上述2個階段運行成功之后，再在集群上對更多的數(shù)據(jù)進行測試.在這一過程中重點關(guān)注作業(yè)的運算性能和資源使用情況，可以利用MapReduce的計數(shù)器功能查看系統(tǒng)及用戶自定義的計數(shù)器，從而優(yōu)化作業(yè)配置.

上述開發(fā)流程將問題以最小代價暴露出來，充分利用單機調(diào)試的便利性，盡量減少集群調(diào)試的需要，整體上降低了開發(fā)者定位問題的難度，有效地提高了開發(fā)效率.

5.5 MapReduce作業(yè)調(diào)優(yōu)

MapReduce程序開發(fā)者除了要保證數(shù)據(jù)處理邏輯的正確性之外，還需要關(guān)注作業(yè)在集群中的運行性能和資源消耗.后者要求開發(fā)者對數(shù)據(jù)處理邏輯以及MapReduce和YARN系統(tǒng)的細(xì)節(jié)有深入的了解，能夠根據(jù)實際情況調(diào)優(yōu)作業(yè)參數(shù)，這無疑增加了MapReduce用戶的使用成本.在協(xié)助用戶進行作業(yè)性能分析和參數(shù)優(yōu)化的過程中，發(fā)現(xiàn)常見的問題可以按處理階段概括為以下3類：

1) Map階段.內(nèi)存配置不合理導(dǎo)致內(nèi)存數(shù)據(jù)頻繁落地磁盤，磁盤IO開銷大.

2) Shuffle階段.Map輸出未壓縮導(dǎo)致Shuffle數(shù)據(jù)量過大，帶寬開銷大；Reduce端的Shuffle內(nèi)存及并發(fā)參數(shù)的配置不合理導(dǎo)致磁盤IO開銷大或數(shù)據(jù)拉取慢.

3) Reduce階段.任務(wù)并發(fā)數(shù)不足導(dǎo)致單任務(wù)處理數(shù)據(jù)量過大；Reduce的輸出數(shù)據(jù)過大和HDFS多副本導(dǎo)致帶寬開銷大等.

上述這些問題覆蓋了實際應(yīng)用中大部分的性能調(diào)優(yōu)的場景.為了減少用戶的使用門檻，可以利用Hadoop系統(tǒng)為每個作業(yè)記錄歷史文件，分析其中的任務(wù)數(shù)和各種系統(tǒng)計數(shù)器，判斷可能的參數(shù)優(yōu)化點，再提醒用戶去關(guān)注相關(guān)問題.這種自動化的流程也有效地降低了集群的運營成本.例如在實踐中曾遇到某一作業(yè)，雖然能夠正常運行，但整體運行比同規(guī)模作業(yè)時間長很多.通過自動化分析，發(fā)現(xiàn)問題在于其Map階段Java GC時間占比很大(用戶的Map算法頻繁利用內(nèi)存進行數(shù)據(jù)緩存)，因此本平臺調(diào)大了Map階段的內(nèi)存需求量，從而使單Map任務(wù)時間減少為原來的15，作業(yè)整體時間也大幅縮短.表5是優(yōu)化前后的CPU耗時對比.

Fig. 4 Architecture of Minos deployment system圖4 Minos部署系統(tǒng)架構(gòu)圖

CategoryCPUTime∕msGCTime∕msGCTimeoverCPUTime∕%BeforeOptimization74686047201563．20AfterOptimization12956210270．79

6 集群管理

隨著接入業(yè)務(wù)數(shù)量的增加和集群規(guī)模的增長，集群的布署、升級、監(jiān)控以及管理成為了一個挑戰(zhàn)，亟需一套能夠方便布署、升級集群，同時能夠直觀查看集群運行狀態(tài)的系統(tǒng).希望能夠通過這樣的系統(tǒng)，一方面可以降低集群維護成本，減輕維護集群的壓力；另一方面可以實時查看集群的運行狀態(tài)，讓團隊成員和用戶了解集群的健康狀況，同時也可以及時把集群的故障反饋給團隊成員，能夠讓團隊成員在第一時間發(fā)現(xiàn)問題、解決問題，把對業(yè)務(wù)的影響降到最小.

業(yè)內(nèi)已有的解決方案，包括Hadoop原生的布署腳本、Cloudera Manager[32]和Apache Ambari[33]等盡管有各自的優(yōu)點與缺點，但都與本文要研究的目標(biāo)系統(tǒng)有一些距離.因此本文提出了一套自主設(shè)計和實現(xiàn)的Hadoop布署和監(jiān)控系統(tǒng)Minos，目前該系統(tǒng)已經(jīng)開源[34].圖4是Minos部署系統(tǒng)的架構(gòu)圖，整體系統(tǒng)主要由4個組件組成.

1) 客戶端(client).直接提供給用戶使用的命令行工具.用戶可以用來部署和管理多種系統(tǒng)的集群服務(wù)與進程，包括安裝、啟停、清除等.

2) 監(jiān)控面板(owl).展示集群服務(wù)和進程狀態(tài)的網(wǎng)站.它通過JMX[35]接口從它管理的各個進程收集內(nèi)部數(shù)據(jù)和狀態(tài)，并根據(jù)集群的配置，按照服務(wù)、作業(yè)、任務(wù)(ServiceJobTask)3個級別匯總和展示.

3) 監(jiān)視進程(supervisor).部署在集群的所有機器上，負(fù)責(zé)管理和監(jiān)控服務(wù)的所有進程.Supervisor原本是一個開源項目[36]，提供了一套讓用戶在類UNIX操作系統(tǒng)上遠(yuǎn)程監(jiān)控和控制進程的方法.本文根據(jù)Minos的需要進行了擴展和改進，主要增加了一套RPC接口供Minos Client調(diào)用.

4) 包管理服務(wù)器(tank).集群運行所使用的軟件包集中管理和存放的服務(wù)器.Minos以包名和版本號來唯一表示一個軟件包.

使用Minos系統(tǒng)部署和管理一個集群服務(wù)的典型流程如下：

1) 安裝Minos系統(tǒng)(所有集群服務(wù)僅需要做一次)，安裝集群服務(wù)所需要的軟件包到Tank;

2) 編寫集群配置文件，通過Minos Client初始化集群;

3) 查看集群運行狀態(tài)，根據(jù)需求啟停、更新、清除集群服務(wù).

Minos系統(tǒng)已經(jīng)成為內(nèi)部部署和管理大數(shù)據(jù)平臺各個組件服務(wù)的標(biāo)準(zhǔn)工具，目前支持了在使用的主流開源系統(tǒng)，包括Hadoop(HDFSYARN)，ZooKeeper，HBase，Impala，Storm等.它大大降低了管理和維護這些大規(guī)模分布式系統(tǒng)的成本，提升了業(yè)務(wù)團隊的生產(chǎn)效率.根據(jù)實際使用的經(jīng)驗，Minos系統(tǒng)主要具有6個特點:

1) 提供了直觀的Web界面來查看集群的運行狀態(tài)， 提供了命令行工具來管理集群，方便快速定位錯誤.

2) 放寬了布署服務(wù)必須是系統(tǒng)級服務(wù)的約束，支持同機運行多個實例.這個特性主要的應(yīng)用場景是在大內(nèi)存的機器上通過布署多個RegionServer來提高機器內(nèi)存的使用率，同時能避免單個RegionServer的堆太大而導(dǎo)致的GC時間過長引起的一系列問題.

3) 靈活的包管理功能，對開發(fā)團隊更加友好.這個特性主要的好處有：①對于同一個系統(tǒng)特定的版本，團隊內(nèi)部只要有一位成員構(gòu)建，其他成員便可以方便地復(fù)用編譯好的軟件包；②對于同一個系統(tǒng)不同版本的軟件包都有明確的標(biāo)識，互相不影響；③所有軟件包都集中管理，有直觀的Web界面進行操作.

4) 在集群中抽象出了ServiceJobTask的概念，能夠通過配置文件直觀、簡潔地描述集群.

5) 對集群的管理既支持集群級別的管理，也支持JobTask級別的管理.這個特性可以靈活地支持操作整個集群，或者是集群中的某些JobTask.

6) 監(jiān)控指標(biāo)的收集與展示采用了OpenTSDB[37], 具有強大的線型擴展性.由于Hadoop系統(tǒng)的監(jiān)控指標(biāo)較多，需要存儲的時間較長，在前期采用MySQL來存儲這些指標(biāo)時，隨著集群規(guī)模的增長，很快MySQL就成為了瓶頸.后來經(jīng)過調(diào)研，本平臺把MySQL換成了OpenTSDB，由于OpenTSDB底層的存儲是基于HBase的，HBase本身具有強大的線型擴展性，因此Minos中指標(biāo)存儲的問題便得到了很好的解決.

很多業(yè)務(wù)已經(jīng)接入或正在接入本平臺的存儲與計算集群.目前，整體數(shù)據(jù)存儲量已達到PB級規(guī)模，每天運行計算作業(yè)2 000多個，吞吐量在50TB左右.圖5展示了2013年8月至11月的每日作業(yè)數(shù)情況.

Fig. 5 Daily running jobs of MapReduce圖5 MapReduce每日作業(yè)數(shù)

7 未來工作

7.1 計算系統(tǒng)

Hadoop YARN平臺在支持現(xiàn)有MapReduce計算的同時，也為未來更多的擴展成為可能.目前很多開源項目支持在YARN平臺上運行或部署，包括Storm[20]，Spark[21]，Tez[38]，Impala[22].這些項目擴展了分布式計算模型，對特定領(lǐng)域有更好的支持.本文也嘗試將這些項目應(yīng)用到計算集群上，在復(fù)用集群的同時為用戶提供更多的選擇.此外，YARN也有發(fā)展成為通用部署平臺的潛力，目前已經(jīng)有將HBase部署在YARN上的開源項目，我們也會在這一領(lǐng)域繼續(xù)探索和嘗試.

7.2 存儲系統(tǒng)

HDFS目前已經(jīng)基本能夠滿足大部分業(yè)務(wù)的需求，但是隨著業(yè)務(wù)規(guī)模的增長，也凸顯出一些新的需求.此外HDFS本身的易用性方面也有很大的提高空間，未來的5個主要發(fā)展方向如下：

1) 名字服務(wù).支持通過名字訪問HDFS集群.

2) HDFS Raid.希望在減少備份數(shù)的同時不損失數(shù)據(jù)的可靠性，從而達到節(jié)約成本的目的.

3) HDFS QoS.希望能夠?qū)τ脩籼峁┑姆?wù)有基本的網(wǎng)絡(luò)延遲和吞吐量的保證，同時保障數(shù)據(jù)的可靠.

4) 冷熱數(shù)據(jù)分離.希望對冷熱數(shù)據(jù)使用不用的策略和備份數(shù)，進一步降低存儲成本.

5) 跨數(shù)據(jù)中心同步.

7.3 集群管理

本文的數(shù)據(jù)存儲與計算平臺主要基于開源系統(tǒng).在受益于開源系統(tǒng)提供便利的同時，也希望能做一些事情來回饋開源社區(qū)，這是把Minos開源出去的主要目的.另外也希望能夠借助社區(qū)的力量，一起來完善Minos.當(dāng)前已經(jīng)規(guī)劃要做或者正在做的一些特性主要有：

1) 同機多實例布署的支持；

2) 異構(gòu)機型的支持；

3) 易用性的提升，包括相關(guān)文檔完善、安裝過程自動化等.

7.4 公有云

目前為止，數(shù)據(jù)的收集、存儲、處理、計算平臺都是面向公司內(nèi)部用戶的，屬于私有云的概念.小米公司有提供開放平臺的計劃，把自己擁有的平臺與數(shù)據(jù)開放出去，便于各種應(yīng)用的開發(fā)；同時也會開放數(shù)據(jù)處理的能力，讓更多的用戶收益.在這個場景下會有3個新的挑戰(zhàn)：

1) 多租戶.多個用戶之間是不可見和不相互影響的，需要良好的數(shù)據(jù)和資源隔離來達到這點；同時在多用戶情況下也要達到和用戶約定的服務(wù)等級協(xié)議(service-level agreement, SLA).

2) 安全.因為用戶的數(shù)據(jù)和計算任務(wù)會托管在小米公司提供的環(huán)境里，安全是用戶最為關(guān)心的問題之一.

3) 彈性.用戶的需求是動態(tài)變化的，平臺需要根據(jù)用戶的實際需求來分配資源，以降低用戶的使用成本.

8 總 結(jié)

隨著互聯(lián)網(wǎng)和移動互聯(lián)網(wǎng)的快速發(fā)展和普及，人類所創(chuàng)造的數(shù)據(jù)量和產(chǎn)生的速度都在迅速膨脹，比如用戶訪問日志、用戶生成內(nèi)容(user generated content, UGC)等，客觀上推動了大數(shù)據(jù)問題的研究.大數(shù)據(jù)的一個特點是價值密度較低，但在數(shù)量龐大的數(shù)據(jù)背后，隱藏著深刻的規(guī)律和洞見.對這些規(guī)律的挖掘和發(fā)現(xiàn)，一方面可以為企業(yè)帶來巨大的商業(yè)價值，獲得超越其他競爭對手的優(yōu)勢；另一方面也能豐富用戶服務(wù)，提供更穩(wěn)定、更優(yōu)異的使用體驗.因此，如何從這些龐大、分散的數(shù)據(jù)中去粗存精，沙里淘金，是大數(shù)據(jù)要解決的問題和面臨的挑戰(zhàn).

本文從小米公司的行業(yè)應(yīng)用和實踐出發(fā)，在深入研究現(xiàn)有平臺的基礎(chǔ)上，提出了一種基于開源生態(tài)系統(tǒng)的大數(shù)據(jù)收集與處理平臺的設(shè)計方案.同時針對現(xiàn)有開源軟件在功能、一致性、可用性和效率等關(guān)鍵問題上的缺陷，提出了相應(yīng)的優(yōu)化和改進方案，并在業(yè)務(wù)系統(tǒng)中得以實施和驗證.

當(dāng)然，本文提出的大數(shù)據(jù)平臺還有需要改進和完善的地方，比如計算模型較為單一、存儲尚未支持冷熱數(shù)據(jù)分離、尚未提供跨數(shù)據(jù)中心的同步功能等.下一步研究工作將集中在全面的計算模型、低成本存儲、跨數(shù)據(jù)中心同步、多租戶等問題上.

[1]Zikopoulos P, Eaton C. Understanding Big Data: Analytics for Enterprise Class Hadoop and Streaming Data[M]. New York: McGraw-Hill, 2011

[2]Slee M, Agarwal A, Kwiatkowski M. Thrift: Scalable cross-language services implementation[ROL]. Palo Alto: Facebook, 2007 [2015-06-08]. https:thrift.apache.orgstaticfilesthrift-20070401.pdf

[3]Shao Z. Real-time analytics at Facebook[C]Proc of the 5th Extremely Large Databases Conf. Menlo Park: SLAC National Accelerator Laboratory, 2011: 21-33

[4]Shao Z. Real-time analytics at Facebook: Data freeway and puma[COL]Proc of 2011 Hadoop in China. [2015-04-18]. http:hic2011.hadooper.cndctattachY2xiOmNsYjpwZGY6MTQxMzY=

[5]Facebook. Scribe[CPOL]. [2015-06-08]. https:github.comfacebookscribe

[6]Apache. Hadoop[CPOL]. [2015-06-08]. http:hadoop.apache.org

[7]Dean J, Ghemawat S. MapReduce: Simplified data processing on large clusters[J]. Communications of the ACM, 2008, 51(1): 107-113

[8]Apache. Hive[CPOL]. [2015-06-08]. http:hive.apache.org

[9]Apache. HBase[CPOL]. [2015-06-08]. http:hbase.apache.org

[10]Cheng Miao, Chen Huaping. Weblog mining based on Hadoop[J]. Computer Engineering, 2011, 37(11): 37-39 (in Chinese)(程苗, 陳華平. 基于Hadoop的Web日志挖掘[J]. 計算機工程, 2011, 37(11): 37-39)

[11]Song Ying, Shen Qiwei, Wang Jing. Design and implementation of Web log pre-processing based on Hadoop[J]. Telecom Engineering Technics and Standardization, 2011, 24(11): 84-89 (in Chinese)(宋瑩, 沈奇威, 王晶. 基于Hadoop的Web日志預(yù)處理的設(shè)計與實現(xiàn)[J]. 電信工程技術(shù)與標(biāo)準(zhǔn)化, 2011, 24(11): 84-89)

[12]Liu Yongzeng, Zhang Xiaojing, Li Xianyi. Design of Web log analysis system based on HadoopHive[J]. Journal of Guangxi University: Natural Science Edition, 2011, 36(Suppl1): 314-317 (in Chinese)(劉永增, 張曉景, 李先毅. 基于HadoopHive的Web日志分析系統(tǒng)的設(shè)計[J]. 廣西大學(xué)學(xué)報: 自然科學(xué)版, 2011, 36(增刊1): 314-317)

[13]Zhu Zhu. Research and application of massive data processing model based on Hadoop[D]. Beijing: Beijing University of Posts and Telecommunications, 2008 (in Chinese)(朱珠. 基于Hadoop的海量數(shù)據(jù)處理模型研究和應(yīng)用[D]. 北京: 北京郵電大學(xué), 2008)

[14]Li Jun. Exploration on the cloud computing model based on Hadoop[J]. Information Security and Technology, 2011 (6): 30-32 (in Chinese)(李珺. 基于Hadoop云計算模型探究[J]. 信息安全與技術(shù), 2011 (6): 30-32)

[15]Wan Zhizhen. Design and implementation of parallel computing platform based on MapReduce model[D]. Hangzhou: Zhejiang University, 2008 (in Chinese)(萬至臻. 基于MapReduce模型的并行計算平臺的設(shè)計與實現(xiàn)[D]. 杭州: 浙江大學(xué), 2008)

[16]Cui Jie, Li Taoshen, Lan Hongxing. Design and development of the mass data storage platform based on Hadoop[J]. Journal of Computer Research and Development, 2012, 49(Suppl1): 12-18 (in Chinese)(崔杰, 李陶深, 蘭紅星. 基于Hadoop的海量數(shù)據(jù)存儲平臺設(shè)計與開發(fā)[J]. 計算機研究與發(fā)展, 2012, 49(增刊1): 12-18)

[17]Dong He, Xu Lingyu. SaaS-Flow system structure based on cloud platform[J]. Journal of Shanghai University: Natural Science Edition , 2013, 19(1): 14-20 (in Chinese)(董賀, 徐凌宇. 基于云平臺的軟件服務(wù)流體系結(jié)構(gòu)[J]. 上海大學(xué)學(xué)報：自然科學(xué)版, 2013, 19(1): 14-20)

[18]Ji Jun. Design and implementation of a data mining platform architecture based on cloud computing[D]. Qingdao: Qingdao University, 2009 (in Chinese)(紀(jì)俊. 一種基于云計算的數(shù)據(jù)挖掘平臺架構(gòu)設(shè)計與實現(xiàn)[D]. 青島: 青島大學(xué), 2009)

[19]Hunt P, Konar M, Junqueira F P, et al. ZooKeeper: Wait-free coordination for Internet-scale systems[C]Proc of the 2010 USENIX Annual Technical Conf. Berkeley: USENIX Association, 2010: 11-18

[20]Apache. Storm[CPOL]. [2015-06-08]. http:storm.apache.org

[21]Apache. Spark[CPOL]. [2015-06-08]. http:spark.incubator.apache.org

[22]Cloudera. Impala[CPOL]. [2015-06-08]. http:impala.io

[23]Google. Protocol Buffer[CPOL]. [2015-06-08]. https:code.google.compprotobuf

[24]Apache. Kafka[CPOL]. [2015-06-08]. https:kafka.apache.org

[25]Apache. Flume[CPOL]. [2015-06-08]. http:flume.apache.org

[26]Apache. Chukwa[CPOL]. [2015-06-08]. http:chukwa.apache.org

[27]Google. Snappy[CPOL]. [2015-06-08]. http:google.github.iosnappy

[28]Ghemawat S, Gobioff H, Leung S T. The Google file system[C]Proc of the 19th ACM Symp on Operating Systems Principles. New York: ACM, 2003: 29-43

[29]Apache. HDFS-4660[CPOL]. [2015-06-08]. https:issues.apache.orgjirabrowseHDFS-4660

[30]Pike R, Dorward S, Griesemer R, et al. Interpreting the data: Parallel analysis with Sawzall[J]. Scientific Programming, 2005, 13(4): 277-298

[31]Melnik S, Gubarev A, Long J J, et al. Dremel: Interactive analysis of Web-scale datasets[J]. Proceedings of the VLDB Endowment, 2010, 3(12): 330-339

[32]Cloudera. Cloudera Manager[CPOL]. [2015-06-08]. https:www.cloudera.comproductscloudera-manager.html

[33]Apache. Ambari[CPOL]. [2015-06-08]. http:ambari.apache.org

[34]Xiaomi. Minos[CPOL]. [2015-06-08]. https:github.comXiaoMiminos

[35]Oracle. JMX:[CPOL]. [2015-06-08]. http:www.oracle.comtechnetworkarticlesjavajavamanagement-140525.html

[36]Agendaless Consulting and Contributors. Supervisor[CPOL]. [2015-06-08]. http:supervisord.org

[37]StumbleUpon. OpenTSDB[CPOL]. [2015-06-08]. http:

[38]Apache. Tez[CPOL]. [2015-06-08]. http:tez.incubator.apache.org

Lei Jun, born in 1969. PhD candidate. Founder, board chairman and CEO of Xiaomi Inc. His main research interests include software engineering, distributed system, storage system, big data and high performance computing.

Ye Hangjun, born in 1976. PhD. Software engineer of Xiaomi Inc. His main research interests include distributed system, storage system and cloud computing (yehangjun@xiaomi.com).

Wu Zesheng, born in 1986. Bachelor. Former software engineer of Xiaomi Inc and co-founder of Hangzhou Bongmi Technology Co, Ltd. His main research interests include distributed system and cloud computing (wuzesheng@bongmi.com).

Zhang Peng, born in 1984. Master. Software engineer of Xiaomi Inc. His main research interests include distributed computing system and resource management system (peng.zhang@xiaomi.com).

Xie Long, born in 1984. Master. Software engineer of Xiaomi Inc. His main research interests include high availability and high performance in distributed system (xielong.me@gmail.com).

He Yanxiang, born in 1952. PhD, professor and PhD supervisor. Member of China Computer Federation. His main research interests include trusted software, distributed parallel processing and high performance computing.

Big-Data Platform Based on Open Source Ecosystem

Lei Jun1,2, Ye Hangjun2, Wu Zesheng2, Zhang Peng2, Xie Long2, and He Yanxiang1,3

1(ComputerSchool,WuhanUniversity,Wuhan430072)2(XiaomiInc,Beijing100085)3(StateKeyLaboratoryofSoftwareEngineering(WuhanUniversity),Wuhan430072)

As large-scale data collecting and processing are being widely studied in recent years, several released big data processing platforms are increasingly playing important roles in the operations of many Internet businesses. Open source ecosystems, the engine of big data innovation, have been evolving so rapidly that a number of them are successfully adopted as the components of mainstream data processing platforms. In reality, however, the open source software is still far from perfect while dealing with real large-scale data. On the basis of the industrial practice at Xiaomi Inc, this paper proposes an improved platform for collecting and processing large-scale data in face of varied business requirements. We focus on the problems in terms of the functionality, consistency and availability of the software when they are executed for data collecting, storing and processing procedures. In addition, we propose a series of optimizations aiming at load balance, failover, data compression and multi-dimensional scheduling to significantly improve the efficiency of the current system. All these designs and optimizations described in this paper have been practically implemented and deployed to support various Internet services provided by Xiaomi Inc.

Hadoop; open source ecosystem; big data; data center; network virtualization

2015-06-12；

2016-08-08

國家自然科學(xué)基金項目(91118003，61373039，61170022) This work was supported by the National Natural Science Foundation of China (91118003, 61373039, 61170022).

TP391