王永坤，金耀輝,2+

1.上海交通大學(xué) 網(wǎng)絡(luò)信息中心，上海 200240

2.上海交通大學(xué) 光纖通信國家重點(diǎn)實(shí)驗(yàn)室，上海 200240

1 引言

大數(shù)據(jù)研究及應(yīng)用在最近幾年得到了快速發(fā)展。大數(shù)據(jù)對各個(gè)行業(yè)的運(yùn)行方式都產(chǎn)生了重大影響。在行政方面，大數(shù)據(jù)利用民生數(shù)據(jù)等幫助提高治理水平；在產(chǎn)業(yè)方面，大數(shù)據(jù)在需求、物流、營銷等各個(gè)環(huán)節(jié)提供優(yōu)化建議；在科學(xué)研究方面，大數(shù)據(jù)加速科學(xué)數(shù)據(jù)分析，幫助發(fā)現(xiàn)新領(lǐng)域。國家也對大數(shù)據(jù)技術(shù)高度重視?！笆濉币?guī)劃中提出了“國家大數(shù)據(jù)戰(zhàn)略”。國務(wù)院印發(fā)了《促進(jìn)大數(shù)據(jù)發(fā)展行動綱要》，推動形成公共信息資源共享共用和大數(shù)據(jù)產(chǎn)業(yè)健康發(fā)展的良好格局。以往塵封的數(shù)據(jù)都在國家政策指導(dǎo)下逐步開放，特別是城市數(shù)據(jù)，關(guān)系民生，價(jià)值巨大，推動了智慧城市的發(fā)展。

經(jīng)過近幾年的研究和實(shí)踐，大數(shù)據(jù)的很多理論和系統(tǒng)架構(gòu)都逐漸成熟。但是在實(shí)際應(yīng)用中，仍然有很多挑戰(zhàn)。例如城市的數(shù)據(jù)主管部門可能缺乏大數(shù)據(jù)的存儲和計(jì)算的設(shè)施；數(shù)據(jù)分析平臺無法最大限度地共享數(shù)據(jù)的同時(shí)又隔離有害代碼，保證數(shù)據(jù)和計(jì)算平臺的安全等。

已有很多研究和系統(tǒng)在解決這些問題。開源社區(qū)提供了很多開源大數(shù)據(jù)解決方案，例如Apache Hadoop、Apache Hive[1]、Apache Spark[2]，來解決數(shù)據(jù)的存儲、查詢和計(jì)算，并且有專門公司提供企業(yè)級服務(wù)。在數(shù)據(jù)平臺之上，提供了HUE、Jupyter、Zepplin等交互計(jì)算接口，并提供基本的訪問控制。在學(xué)術(shù)界，DataHub[3]是MIT開發(fā)的一個(gè)試驗(yàn)性的平臺，提供了數(shù)據(jù)的存儲、共享和協(xié)作服務(wù)；用戶可以發(fā)布自己的數(shù)據(jù)，也可以使用別人的數(shù)據(jù)。數(shù)據(jù)集有版本。用戶可以通過Thrift使用不同的語言來操作數(shù)據(jù)。俄勒岡健康與科學(xué)大學(xué)（Oregon Health and Science University）與英特爾公司（Intel）合作建立了一個(gè)精準(zhǔn)醫(yī)療分析平臺，用于共享分析各機(jī)構(gòu)的醫(yī)療數(shù)據(jù)，同時(shí)保護(hù)患者的隱私1）http://www.ohsu.edu/blogs/96kmiles/2015/08/20/intel-ohsu-announce-collaborative-cancer-cloud-at-intel-developer-forum/。。工業(yè)界有很多建立在云上的大數(shù)據(jù)平臺方案，包括國外的亞馬遜云AWS（Amazon Web services）、谷歌云平臺大數(shù)據(jù)產(chǎn)品、微軟云（Azure）上的Hadoop，國內(nèi)有阿里云大數(shù)據(jù)服務(wù)ODPS（open data processing service）、天池及數(shù)加，騰訊大數(shù)據(jù)，百度大數(shù)據(jù)+。

上面提到的方案中，學(xué)術(shù)界的新想法需要更多實(shí)踐來驗(yàn)證；工業(yè)界的方案規(guī)模大，技術(shù)水平高，相對而言成本也比學(xué)校等非盈利機(jī)構(gòu)的平臺要高，而且涉及的知識產(chǎn)權(quán)問題等較嚴(yán)格，數(shù)據(jù)開放共享的難度較大。

因此，本文的目的是完全獨(dú)立地使用開源社區(qū)的解決方案來搭建一個(gè)一站式的共享數(shù)據(jù)、計(jì)算和代碼的數(shù)據(jù)平臺。本文的平臺完全使用開源軟件，自己選取設(shè)計(jì)組件，自己搭建和運(yùn)維。開源軟件代碼公開并且由開源社區(qū)維護(hù)，非常適合高校這種IT經(jīng)費(fèi)相對較少但是智力資源較多的環(huán)境。本文的平臺用于校內(nèi)及部分公開服務(wù)，也定期提供給數(shù)據(jù)大賽這種大規(guī)模、高強(qiáng)度、集中式、密集計(jì)算的場景使用。根據(jù)了解，這在國內(nèi)外高校中也是一種大膽的嘗試，希望該經(jīng)驗(yàn)可以給其他院校和機(jī)構(gòu)一些參考。

本文的貢獻(xiàn)概括如下：

（1）設(shè)計(jì)了一種數(shù)據(jù)和計(jì)算共享的數(shù)據(jù)平臺架構(gòu)，包含了數(shù)據(jù)平臺和代碼倉庫，代碼可以自動調(diào)度到平臺上運(yùn)行；

（2）選取硬件并配置了生產(chǎn)環(huán)境，給出基本的評測結(jié)果驗(yàn)證了這種設(shè)計(jì)的可行性；

（3）使用這個(gè)數(shù)據(jù)平臺服務(wù)了大規(guī)模的上海開放數(shù)據(jù)創(chuàng)新應(yīng)用大賽，介紹了其中的經(jīng)驗(yàn)和見解。

本文內(nèi)容組織如下：第2章介紹問題背景及需求；第3章介紹系統(tǒng)架構(gòu)設(shè)計(jì)；第4章介紹系統(tǒng)的基本評估；第5章介紹數(shù)據(jù)平臺在上海開放數(shù)據(jù)創(chuàng)新應(yīng)用大賽中的應(yīng)用；第6章總結(jié)全文。

2 背景及需求

2.1 異構(gòu)數(shù)據(jù)實(shí)時(shí)采集和存儲

得益于各行各業(yè)信息化的普及和物聯(lián)網(wǎng)的發(fā)展，數(shù)據(jù)量正以驚人的速度增長。同時(shí)因?yàn)閿?shù)據(jù)源不同，所以數(shù)據(jù)格式也多種多樣，呈現(xiàn)異構(gòu)特征。如何采集、傳輸和存儲大量異構(gòu)的數(shù)據(jù)是一個(gè)非常大的挑戰(zhàn)。

采集程序通常分為實(shí)時(shí)采集和批量采集。實(shí)時(shí)采集的程序直接和生產(chǎn)環(huán)境的數(shù)據(jù)源對接，將最新生成的數(shù)據(jù)直接傳回。一般采集程序是非常輕量級的，不能影響生產(chǎn)環(huán)境的運(yùn)行，并且可以斷點(diǎn)續(xù)傳。批量采集程序一般是針對離線庫或者備份定期進(jìn)行大量數(shù)據(jù)回傳，對生產(chǎn)環(huán)境的影響較小。

數(shù)據(jù)的可靠傳輸可以保證數(shù)據(jù)不丟包，最后的運(yùn)算結(jié)果不出現(xiàn)偏差。一般的做法是針對每個(gè)或者多個(gè)數(shù)據(jù)包，接收端采取回復(fù)確認(rèn)包（ACK）的方式來確保數(shù)據(jù)正確收到。沒有收到的數(shù)據(jù)會要求重傳。ACK的管理可以是集中式或者分布式。集中式的回傳確認(rèn)方式是發(fā)送方將已發(fā)數(shù)據(jù)包編號發(fā)給全局控制節(jié)點(diǎn)，接收方收到包后也將確認(rèn)包發(fā)給同一個(gè)控制節(jié)點(diǎn)，控制節(jié)點(diǎn)來控制是否重發(fā)。優(yōu)點(diǎn)是實(shí)現(xiàn)非常簡單；缺點(diǎn)是控制節(jié)點(diǎn)可能成為性能瓶頸，并且有可能成為單點(diǎn)故障點(diǎn)（single point of failure，SPOF）。比較魯棒的做法是讓每個(gè)節(jié)點(diǎn)維護(hù)已發(fā)送包列表和確認(rèn)包列表，將確認(rèn)包一直回傳至源發(fā)送節(jié)點(diǎn)。這種分布式的實(shí)現(xiàn)方式吞吐量大，整個(gè)系統(tǒng)可靠性高，但是實(shí)現(xiàn)復(fù)雜。第三種方式是使用集中的分布式存儲來緩存數(shù)據(jù)，發(fā)送方和訂閱方可以在一跳內(nèi)得到數(shù)據(jù)。這種方式需要分布式存儲，并且發(fā)送方和訂閱方一般需要在同一子網(wǎng)內(nèi)。本文采用第三種方式，利用Kafka來實(shí)現(xiàn)數(shù)據(jù)的傳輸和短期分享。

不同的數(shù)據(jù)源會生成不同的數(shù)據(jù)，形成多維數(shù)據(jù)，數(shù)據(jù)字段和格式不相同。對有些數(shù)據(jù)如交換機(jī)流量甚至需要自己定義如何從數(shù)據(jù)流中分割出數(shù)據(jù)包。由于數(shù)據(jù)生成速度快而且數(shù)據(jù)量較大，不適合使用傳統(tǒng)的文件系統(tǒng)和關(guān)系型數(shù)據(jù)庫，需利用分布式的文件系統(tǒng)來進(jìn)行實(shí)時(shí)存儲，并且選取的分布式文件系統(tǒng)需要很好地支持后續(xù)的計(jì)算。例如谷歌設(shè)計(jì)了GFS（Google file system）[4]來進(jìn)行分布式存儲。開源社區(qū)也開發(fā)了HDFS（Hadoop distributed file system），廣泛用于各個(gè)大數(shù)據(jù)存儲處理場景。目前流行的企業(yè)內(nèi)部的數(shù)據(jù)湖（data lake）的做法也一般使用HDFS來進(jìn)行存儲。

2.2 異構(gòu)數(shù)據(jù)的計(jì)算

對于多源異構(gòu)數(shù)據(jù)，如果使用傳統(tǒng)的關(guān)系型數(shù)據(jù)庫來處理，需要事先定義好數(shù)據(jù)結(jié)構(gòu)，然后導(dǎo)入數(shù)據(jù)，使用昂貴的硬件來支撐，并且擴(kuò)展性不好，這不適合大量數(shù)據(jù)的場景。因此，谷歌設(shè)計(jì)了名為Big-Table[5]的數(shù)據(jù)模型，變schema-on-write為schema-onread，數(shù)據(jù)可以非常靈活地以動態(tài)的標(biāo)簽或者列名來存儲。谷歌同時(shí)提出了MapReduce[6]的分布式計(jì)算模型，通過使用Map和Reduce操作，可以非常方便地將計(jì)算擴(kuò)展到大量機(jī)器上。

很多傳統(tǒng)的數(shù)據(jù)分析人員仍然偏好使用結(jié)構(gòu)化查詢語言（structured query language，SQL）來進(jìn)行數(shù)據(jù)分析。因此，對于一個(gè)數(shù)據(jù)平臺也需要考慮將SQL轉(zhuǎn)換成分布式的查詢和計(jì)算，降低傳統(tǒng)分析人員的學(xué)習(xí)曲線。目前SQL-on-XXX技術(shù)也比較成熟，比如Apache Hive和Spark SQL等。

2.3 數(shù)據(jù)共享、計(jì)算隔離

大規(guī)模數(shù)據(jù)平臺一般由不同組織、部門的分析人員來共享。即使同一部門的人員，也可能由于業(yè)務(wù)原因，需要對不同業(yè)務(wù)數(shù)據(jù)進(jìn)行訪問控制。因此，數(shù)據(jù)平臺需要細(xì)粒度地支持不同的共享訪問方案；需要支持對數(shù)據(jù)的分層抽象而不需要額外的存儲空間；可以對接和集成常規(guī)的認(rèn)證機(jī)制來保證用戶身份的真實(shí)性；需要支持經(jīng)典的角色訪問控制機(jī)制（role-based access control，RBAC），來給不同業(yè)務(wù)、部門、組織的人員進(jìn)行權(quán)限控制；對于更精細(xì)粒度和更加靈活的訪問控制需求，需要平臺能夠支持屬性訪問控制機(jī)制（attribute-based access control，ABAC）。

對于計(jì)算，需要平臺支持不同的調(diào)度，可以按照各種屬性或者優(yōu)先級來分配資源給不同的任務(wù)。同時(shí)對同等優(yōu)先級的任務(wù)，需要保證公平。

Fig.1 Architecture of data platform圖1 數(shù)據(jù)平臺架構(gòu)及組件

如果數(shù)據(jù)平臺開放給公眾使用，則可能需要對提交的任務(wù)代碼進(jìn)行審核。一般數(shù)據(jù)平臺運(yùn)算的代碼基本都是編譯好的代碼，即使一些腳本語言寫的代碼，在運(yùn)行時(shí)進(jìn)行檢查也比較困難。因此，需要在數(shù)據(jù)平臺集成代碼倉庫的功能，以便按需審核任務(wù)代碼。

3 數(shù)據(jù)平臺架構(gòu)設(shè)計(jì)

本文根據(jù)已有成熟的開源軟件框架，搭建了數(shù)據(jù)平臺。所有組件基本都是開源免費(fèi)的，因?yàn)殚_源軟件代碼公開，并且由開源社區(qū)維護(hù)，非常適合高校這種IT經(jīng)費(fèi)相對較少但是智力資源較多的環(huán)境。本文設(shè)計(jì)的基本架構(gòu)如圖1所示。具體組件及設(shè)計(jì)方法會在下面幾節(jié)展開介紹。

3.1 高可用存儲系統(tǒng)

本文使用Hadoop分布式文件系統(tǒng)（HDFS）來儲存數(shù)據(jù)。HDFS的控制節(jié)點(diǎn)NameNode存儲了文件系統(tǒng)的元數(shù)據(jù)。如果NameNode崩潰，整個(gè)分布式文件系統(tǒng)會停止工作。因此NameNode非常重要，需要提供高可用的熱備功能來保證NameNode常時(shí)在線。Hadoop文件系統(tǒng)早期版本并沒有內(nèi)置熱備（failover）功能，待機(jī)的Secondary NameNode并不能接替崩潰的主NameNode。很多機(jī)構(gòu)或者公司開發(fā)了自己的高可用（high availability，HA）版本。也有一些設(shè)計(jì)依賴外部HA機(jī)制，如DRBD（distributed replicated block device）。直到Hadoop 2.0發(fā)布，內(nèi)置了HA的實(shí)現(xiàn)。默認(rèn)使用的是ZKFC（Zookeeper failover controller），使用Apache Zookeeper集群來進(jìn)行同步鎖。

本文的數(shù)據(jù)平臺也采用默認(rèn)的HA方案。使用兩臺物理節(jié)點(diǎn)（并且部署在不同機(jī)架上）來啟動兩個(gè)NameNode，配置3個(gè)節(jié)點(diǎn)為日志節(jié)點(diǎn)（JournalNode）來在兩個(gè)NameNode之間同步數(shù)據(jù)。使用ZKFC和Zookeeper集群來保證只有一個(gè)活躍NameNode節(jié)點(diǎn)，當(dāng)活躍NameNode無響應(yīng)的時(shí)候主動切換到備用的NameNode。

3.2 高可用作業(yè)調(diào)度系統(tǒng)

本文使用Hadoop YARN作為調(diào)度器，自動調(diào)度和部署用戶的任務(wù)到不同機(jī)器上運(yùn)行。資源管理器（resource manager）是YARN的一個(gè)重要組件，負(fù)責(zé)為所有的任務(wù)調(diào)度資源。因此資源管理器也需要HA配置。本文在兩個(gè)節(jié)點(diǎn)分別配置兩個(gè)一樣的資源管理器，使用內(nèi)置的利用Zookeeper集群的熱備系統(tǒng)來自動檢測失敗的資源管理器，并切換到活躍的資源管理器上。

3.3 安全控制

本文使用Kerberos[7]來驗(yàn)證用戶的身份。Kerberos是一個(gè)“客戶端-服務(wù)器”模式的驗(yàn)證模型。用戶先向認(rèn)證服務(wù)器請求認(rèn)證，獲準(zhǔn)后方可向服務(wù)器請求服務(wù)。

對于權(quán)限控制，使用了RBAC機(jī)制。對文件系統(tǒng)，根據(jù)文件的用戶和分組來進(jìn)行訪問控制；對于Hive查詢，也使用Sentry來實(shí)現(xiàn)訪問控制。

3.4 查詢和計(jì)算

本文的數(shù)據(jù)平臺支持大部分的計(jì)算框架，包括將SQL轉(zhuǎn)換為MapReduce任務(wù)的Apache Hive，使用RDD（resilient distributed datasets）[8]的 Spark 計(jì)算框架，以及原生的MapReduce和Streaming任務(wù)。

Apache Hive仍然是傳統(tǒng)商業(yè)智能（business intelligence，BI）分析人員偏好的工具。可以使用SQL來操作數(shù)據(jù)，易于上手。Spark SQL[9]也類似，使用SQL在Spark計(jì)算框架上方便地進(jìn)行數(shù)據(jù)操作。對于不習(xí)慣命令行的用戶，也提供了HUE（Hadoop user experience）很方便地通過網(wǎng)頁來提交查詢，查看任務(wù)進(jìn)展和查看結(jié)果或出錯(cuò)信息。

對于習(xí)慣于使用IPython[10]Notebook風(fēng)格來進(jìn)行數(shù)據(jù)分析的用戶，也搭建了Jupyter Hub和Zepplin給用戶使用Python Notebook或者Spark Notebook來進(jìn)行交互式的數(shù)據(jù)分析。

另外也開放了數(shù)據(jù)倉庫和商業(yè)智能系統(tǒng)的接口，可以使傳統(tǒng)的數(shù)據(jù)分析工具如Qlik方便地接入。

3.5 數(shù)據(jù)質(zhì)量分析

用戶上傳到數(shù)據(jù)平臺的多源異構(gòu)數(shù)據(jù)通常非常雜亂。即使用戶以為整理好的結(jié)構(gòu)化的數(shù)據(jù)，歷經(jīng)多次數(shù)據(jù)導(dǎo)入、系統(tǒng)升級等操作，實(shí)際也可能存在字段值缺失、不規(guī)范、類型不統(tǒng)一等問題。因此需要工具來幫助用戶方便地查看數(shù)據(jù)質(zhì)量，直觀地給出數(shù)據(jù)質(zhì)量報(bào)告，并且?guī)椭C正不正確的數(shù)據(jù)。

本文使用OpenRefine來幫助用戶快速理解數(shù)據(jù)字段各種數(shù)據(jù)的分布情況，同時(shí)幫助用戶改正、增強(qiáng)相關(guān)數(shù)據(jù)。對于數(shù)據(jù)量比較大的情況，將數(shù)據(jù)統(tǒng)計(jì)分析做成MapReduce任務(wù)，在集群上調(diào)度執(zhí)行。

3.6 數(shù)據(jù)集成

對于如何將數(shù)據(jù)導(dǎo)入到數(shù)據(jù)平臺中，本文使用以下3種方式：

（1）對于流式數(shù)據(jù)，使用Apache Kafka來實(shí)時(shí)進(jìn)行數(shù)據(jù)的分享和交換，同時(shí)將數(shù)據(jù)導(dǎo)入到數(shù)據(jù)平臺的Hadoop中來做離線計(jì)算；

（2）對于結(jié)構(gòu)化數(shù)據(jù)，使用Apache Sqoop來選取相關(guān)的數(shù)據(jù)表，并且盡量按照原來的表結(jié)構(gòu)存儲到數(shù)據(jù)平臺的Hadoop中；

（3）對于其他的各種以文件形式存在的數(shù)據(jù)，在進(jìn)行格式轉(zhuǎn)換后上傳至數(shù)據(jù)平臺中。

3.7 其他組件

本文在數(shù)據(jù)平臺中集成了如下組件：

搜索引擎及查詢可視化系統(tǒng)（Elastic&Kibana）。用戶數(shù)據(jù)被轉(zhuǎn)成JSON（JavaScript object notation）格式，然后存入Elastic集群中并被索引。用戶可以使用Kibana界面來方便地查詢并且實(shí)時(shí)地可視化結(jié)果。

為了滿足一部分用戶的近似實(shí)時(shí)的數(shù)據(jù)存取需求，提供了NoSQL數(shù)據(jù)庫，如Apache Cassandra。Cassandra上層使用的是BigTable的數(shù)據(jù)模型，數(shù)據(jù)存取非常靈活；底層使用的是亞馬遜Dynamo[11]的P2P架構(gòu)，可用性和可靠性非常高。

針對監(jiān)控，部署了Zabbix、Grafana以及Netdata，可以非常方便地查看整個(gè)系統(tǒng)資源的實(shí)時(shí)和歷史利用狀況。

所有的部署和運(yùn)維都是由集群自動化工具Ansible來完成的，重復(fù)部署及擴(kuò)展都很方便。部署和運(yùn)維腳本開發(fā)完成后，可以比較容易地將配置從數(shù)十臺機(jī)器擴(kuò)展到數(shù)百臺甚至更多。

3.8 代碼倉庫、審核和調(diào)度

對于用戶來講，也希望能有一站式的代碼存儲和數(shù)據(jù)查詢的服務(wù)。傳統(tǒng)做法是將代碼提交至Github或者Bitbucket等公有的代碼倉庫，或者企業(yè)、私人的存儲空間中，并將代碼移到數(shù)據(jù)提供方來進(jìn)行查詢，這種方式比較繁瑣。對于數(shù)據(jù)平臺維護(hù)方來說，希望能將有限的資源盡可能有效利用起來，同時(shí)也希望數(shù)據(jù)平臺能免于惡意代碼的攻擊。

針對上述用戶的需求，本文在數(shù)據(jù)平臺中集成了帶版本控制的代碼倉庫的功能，使用GitLab來搭建代碼存儲和版本控制系統(tǒng)。通過GitLab搭建的代碼管理系統(tǒng)，和Github功能類似，可以使用git來管理代碼版本，可以瀏覽代碼，管理缺陷及注釋，并支持團(tuán)隊(duì)協(xié)作開發(fā)。這樣可以幫助平臺運(yùn)維方來掌控用戶的代碼，避免惡意代碼損害平臺和其他用戶的資源。

本文針對用戶的代碼進(jìn)行初步的檢查，幫助用戶提前發(fā)現(xiàn)代碼中遺漏的問題，并且阻止惡意代碼。代碼的自動檢查是個(gè)非常大的挑戰(zhàn)，已有的技術(shù)主要是針對代碼的靜態(tài)分析。Facebook Infer可以檢查Java、C和Objective-C代碼；對于Java，也可以使用經(jīng)典的FindBugs結(jié)合Jenkins，在代碼集成測試的時(shí)候進(jìn)行分析；對于Python代碼可以使用Pylint；對于C和C++代碼，可以使用經(jīng)典的Valgrind來進(jìn)行分析。Cloudera還使用了商業(yè)軟件來對代碼進(jìn)行安全掃描2）Quality Assurance at Cloudera:Static Source-Code Analysis,https://blog.cloudera.com/blog/2016/03/quality-assurance-at-clouderastatic-source-code-analysis/。。

除了掃描用戶的源代碼外，還對數(shù)據(jù)進(jìn)行了細(xì)粒度的訪問控制。給數(shù)據(jù)和帳號仔細(xì)設(shè)置了訪問權(quán)限，最大限度限制了每個(gè)帳號對系統(tǒng)的無關(guān)訪問。并且調(diào)整Hadoop/Spark的資源分配算法，保證在公平的前提下資源盡量得到充分利用。

本文并沒有對用戶的代碼性能進(jìn)行分析和改進(jìn)，但會推薦用戶自己選用合適的數(shù)據(jù)結(jié)構(gòu)和存儲方式來加速計(jì)算，也推薦用戶使用平臺內(nèi)置的工具來幫助提高執(zhí)行速度，比如使用Apache Hive的EXPLAIN命令來查看執(zhí)行計(jì)劃。

本文設(shè)定一些限制條件來選取代碼倉庫的代碼到數(shù)據(jù)平臺運(yùn)行。開設(shè)“發(fā)布（release）”代碼分支倉庫讓用戶將準(zhǔn)備好的代碼提交進(jìn)來。只有源代碼分析通過的任務(wù)才會被調(diào)度到數(shù)據(jù)平臺上執(zhí)行。

3.9 系統(tǒng)備份和恢復(fù)

備份和恢復(fù)是每個(gè)系統(tǒng)必須考慮的問題。本文的數(shù)據(jù)平臺的存儲系統(tǒng)存儲了大量數(shù)據(jù)，對此在應(yīng)用程序?qū)樱℉DFS）默認(rèn)設(shè)置了3個(gè)數(shù)據(jù)副本，這樣既實(shí)現(xiàn)了高可用和高訪問吞吐量，也在一定程度上實(shí)現(xiàn)了數(shù)據(jù)的備份。3個(gè)數(shù)據(jù)副本分別在不同機(jī)架的不同機(jī)器中，避免了機(jī)器或者機(jī)架不可用時(shí)丟失數(shù)據(jù)的情況。元數(shù)據(jù)也同樣重要，本文的數(shù)據(jù)平臺主要有文件系統(tǒng)元數(shù)據(jù)、查詢系統(tǒng)Hive元數(shù)據(jù)。文件系統(tǒng)的元數(shù)據(jù)存儲在控制節(jié)點(diǎn)中，3.1節(jié)介紹了高可用的控制節(jié)點(diǎn)設(shè)計(jì)，兩個(gè)控制節(jié)點(diǎn)互為備份。因?yàn)樵獢?shù)據(jù)總量較小，所以也在考慮使用離線備份工具備份并保存最近的數(shù)據(jù)。查詢系統(tǒng)Hive元數(shù)據(jù)保存在關(guān)系型數(shù)據(jù)庫MySQL中。本文設(shè)計(jì)了MySQL集群來保證高可用性，同時(shí)也做定期的備份。

代碼倉庫GitLab也使用文件系統(tǒng)存儲數(shù)據(jù)，使用關(guān)系型數(shù)據(jù)庫來存儲元數(shù)據(jù)。本文參考了官方文檔中的高可用設(shè)計(jì)3）GitLab Documentation HighAvailability,http://docs.gitlab.com/ce/administration/high_availability/README.html。：使用負(fù)載均衡、網(wǎng)絡(luò)文件系統(tǒng)（network file system，NFS）等來保證高可用性。對于元數(shù)據(jù)，也使用了數(shù)據(jù)庫集群，并且做定期備份。

3.10 系統(tǒng)擴(kuò)展

本文分析了未來的潛在需求，數(shù)據(jù)平臺的系統(tǒng)擴(kuò)展主要在存儲和計(jì)算上。歸功于Hadoop等軟件的高可擴(kuò)展性設(shè)計(jì)，數(shù)據(jù)平臺的線性擴(kuò)展非常容易。新添加的服務(wù)器可以快速安裝Hadoop的計(jì)算和存儲模塊，加入集群參與存儲和計(jì)算。本文使用的Hadoop版本也實(shí)現(xiàn)了HDFS Federation的功能，因此文件系統(tǒng)元數(shù)據(jù)也可以通過給控制節(jié)點(diǎn)添加服務(wù)器來進(jìn)行線性擴(kuò)展。

4 應(yīng)用評估

下面介紹如何根據(jù)需求選取服務(wù)器來搭建生產(chǎn)環(huán)境，并進(jìn)行基本的評估。

4.1 平臺部署

本文簡單地將服務(wù)器按功能分為兩大類：一類是控制節(jié)點(diǎn)；另一類是計(jì)算和存儲節(jié)點(diǎn)?？刂乒?jié)點(diǎn)需要存取元數(shù)據(jù)，對內(nèi)存和存儲性能要求較高。因此使用大內(nèi)存節(jié)點(diǎn)（256 GB），存儲介質(zhì)使用了多塊高性能、大容量的閃存固態(tài)盤（solid state drives，SSD），這些固態(tài)盤配置成磁盤陣列RAID10模式（磁盤組內(nèi)是RAID1，磁盤組間是RAID0），容錯(cuò)能力比較好。計(jì)算和存儲節(jié)點(diǎn)的存儲容量要求較高，因此給每個(gè)節(jié)點(diǎn)選用了12塊6 TB的磁盤，并且配置成JBOD（just a bunch of disks）模式，是為了更好地利用磁盤空間。對于數(shù)據(jù)安全和負(fù)載平衡，則通過數(shù)據(jù)平臺軟件在多機(jī)間實(shí)現(xiàn)。另外對所有節(jié)點(diǎn)都額外配置了2塊10K熱插拔硬盤做成RAID1給操作系統(tǒng)使用。節(jié)點(diǎn)間使用了雙路萬兆網(wǎng)卡互聯(lián)。在操作系統(tǒng)內(nèi)使用動態(tài)鏈路聚合模式將兩路鏈接聚合起來，帶寬加倍可達(dá)到20 Gb/s。集群節(jié)點(diǎn)間的通信帶寬的提高對Hadoop/Spark等的Shuffle操作非常有利，因?yàn)镾huffle操作是Reducer從Mapper那里拷貝中間結(jié)果，涉及大量的網(wǎng)絡(luò)數(shù)據(jù)傳輸。兩種服務(wù)器配置如表1所示。第一期購置了16臺服務(wù)器，包括4臺控制節(jié)點(diǎn)，12臺計(jì)算和存儲節(jié)點(diǎn)，總存儲容量接近900 TB。后期計(jì)劃擴(kuò)容至目前規(guī)模的30倍左右。

Table 1 Specifications of servers表1 數(shù)據(jù)平臺硬件配置

4.2 測試及分析

本節(jié)的目的是驗(yàn)證平臺可以按照預(yù)期來提供服務(wù)。數(shù)據(jù)平臺基本組件都是基于開源軟件和推薦配置，因此本文的目的不是和其他平臺進(jìn)行性能比較，而是對系統(tǒng)進(jìn)行壓力測試，并且提供結(jié)果給用戶來估計(jì)任務(wù)耗時(shí)。有關(guān)數(shù)據(jù)平臺的性能測試，學(xué)術(shù)界、產(chǎn)業(yè)界和開源社區(qū)已經(jīng)有大量很好的工作。有關(guān)測試工具，用戶可以參考中科院計(jì)算所的BigDataBench[12]以及英特爾公司的HiBench[13]。

本文測試的主要軟件部件的版本如表2所示。

Table 2 Software configuration表2 軟件版本

本文使用了Hadoop安裝包內(nèi)置的TeraSort排序任務(wù)來進(jìn)行基本的測試。圖2（a）顯示了本文平臺使用TeraSort對不同大小的數(shù)據(jù)集排序所耗費(fèi)的時(shí)間。圖2（b）顯示了使用不同數(shù)目的Reducer對1 TB數(shù)據(jù)集排序所耗費(fèi)的時(shí)間?？梢钥吹皆诒ＷC結(jié)果正確的前提下，增加Reducer的數(shù)目可以顯著縮短任務(wù)的執(zhí)行時(shí)間。

Fig.2 TeraSort benchmark圖2 TeraSort測試

由于機(jī)器學(xué)習(xí)算法的用戶使用率較高，本文選取了3種典型的算法，分別是用于分類的Bayes、用于聚類的K-Means以及用于排名的PageRank。在Hi-Bench 5.0的默認(rèn)設(shè)置下，對每種算法，分別用Hadoop的Map/Reduce框架和Spark來運(yùn)行測試，結(jié)果如表3所示?？梢钥吹交趦?nèi)存計(jì)算的Spark框架有較大的優(yōu)勢。本測試并非飽和壓力測試，結(jié)果僅供用戶提交任務(wù)時(shí)參考。

本文仍然有很多優(yōu)化的空間，例如使用特別的壓縮文件格式可以顯著提高磁盤輸入輸出（I/O）性能，能夠減少CPU的iowait等待時(shí)間（TeraSort測試中觀察到CPU的一些iowait時(shí)間）。也可以結(jié)合一些特定的易于壓縮的存儲格式（比如Parquet列存儲格式，Parquet部分算法來自Dremel[14]）來研究性能提升。也可以深入分析BigDataBench以及HiBench的壓力測試結(jié)果。

Table 3 HiBench report表3 HiBench測試結(jié)果①

5 開放數(shù)據(jù)大賽SODA應(yīng)用

上海市擁有2 415.27萬4）2015年上海市國民經(jīng)濟(jì)和社會發(fā)展統(tǒng)計(jì)公報(bào)，上海市統(tǒng)計(jì)局，國家統(tǒng)計(jì)局上海調(diào)查總隊(duì)，2016-02-29。常住居民以及龐大的公共設(shè)施系統(tǒng)，如公共交通系統(tǒng)，政府手中積累了大量的與城市生活息息相關(guān)的數(shù)據(jù)。為了提升數(shù)據(jù)的價(jià)值，上海市政府組織了上海開放數(shù)據(jù)創(chuàng)新應(yīng)用大賽（Shanghai Open Data Apps，SODA），利用大眾的智慧來一起挖掘城市數(shù)據(jù)的價(jià)值。大賽總計(jì)有來自國內(nèi)外的2 914人報(bào)名參賽，組隊(duì)817個(gè)。參賽選手的背景非常多樣化，有專業(yè)的數(shù)據(jù)分析師、工程師、設(shè)計(jì)師和產(chǎn)品經(jīng)理等，也有高校教師、公務(wù)員、學(xué)生等。選手來自的機(jī)構(gòu)也包括了國內(nèi)外知名高校（如MIT、清華、交大、復(fù)旦等）和知名企業(yè)（如阿里巴巴）等。大賽征集了有效創(chuàng)意方案總計(jì)505個(gè)5）2015上海開放數(shù)據(jù)創(chuàng)新應(yīng)用大賽決賽鳴鑼，http://sh.qq.com/a/20151116/038896.htm。。數(shù)量眾多的參賽選手和大量的創(chuàng)意方案產(chǎn)生，體現(xiàn)了數(shù)據(jù)的密集使用，離不開對數(shù)據(jù)高強(qiáng)度的計(jì)算。

關(guān)于數(shù)據(jù)，除了上海市政府?dāng)?shù)據(jù)服務(wù)網(wǎng)（http://www.datashanghai.gov.cn）開放的約1 000個(gè)數(shù)據(jù)集外，上海市政府又特別開放了10個(gè)大容量高質(zhì)量的數(shù)據(jù)集，如表4所示。其中一卡通刷卡的數(shù)據(jù)量達(dá)到了4億多條，而出租車行車數(shù)據(jù)更是超過了34億條。本文的設(shè)計(jì)雖然著眼于大規(guī)模的數(shù)據(jù)平臺集群，但是限于硬件資源，搭建的生產(chǎn)環(huán)境仍是一個(gè)小規(guī)模的集群，用于先導(dǎo)試驗(yàn)。這些數(shù)據(jù)對本文設(shè)計(jì)的計(jì)算平臺來講，是一個(gè)極大的挑戰(zhàn)。

Table 4 SODAdataset表4 SODA數(shù)據(jù)集

限于篇幅和大賽組織方對選手代碼和使用行為的保密協(xié)議，本文僅描述一些有趣的規(guī)律，希望能為其他平臺設(shè)計(jì)和優(yōu)化提供參考。

發(fā)現(xiàn)大多數(shù)計(jì)算仍然是用于數(shù)據(jù)統(tǒng)計(jì)的聚集（aggregation）操作。例如，某一個(gè)方案計(jì)算了出租車的平均速度。由于出租車在不同時(shí)間段、不同路段和不同方向上的平均速度是不一樣的，一般通過計(jì)算某一較短時(shí)間段內(nèi)的、某路段上同方向的所有車的平均車速，來設(shè)計(jì)相關(guān)預(yù)測。其中路段是由經(jīng)緯度來按用戶的需求劃分的。下面給出了一個(gè)典型的工作日平均速度計(jì)算的腳本（Apache Hive Query），簡單起見，按小時(shí)來計(jì)算平均值，實(shí)際的時(shí)間間隔要求更短。

上面的代碼非常簡單，但是多人并發(fā)訪問的數(shù)據(jù)量較大。對34億條記錄計(jì)算后結(jié)果會保存到內(nèi)存數(shù)據(jù)庫或者Key-Value Store中，用于近似實(shí)時(shí)的查詢。某些選手只對特別的路段（road_segment）感興趣，會對路段的經(jīng)緯度進(jìn)行過濾。因此路段經(jīng)緯度的索引設(shè)置比較重要，索引方案也要根據(jù)數(shù)據(jù)類型進(jìn)行選取。對路段使用了Hive中簡單穩(wěn)定的Compact-IndexHandler，而對行車方向direction（上行、下行）使用了位圖索引（BitmapIndexHandler）[15]。

Fig.3 Running Pyspark and regression prediction on YARN by Jupyter圖3 使用Jupyter調(diào)用Pyspark在YARN集群上運(yùn)行回歸預(yù)測

發(fā)現(xiàn)大量的聚集操作主要是數(shù)據(jù)預(yù)處理，給后面的分類或預(yù)測算法提供足夠的特征。經(jīng)過預(yù)處理后的數(shù)據(jù)集已經(jīng)變得非常小，因此很多用戶將處理好的數(shù)據(jù)集下載到本地，然后使用算法進(jìn)行分析預(yù)測。例如公交卡刷卡數(shù)據(jù)有4億多條，但是經(jīng)過處理識別后，得到的一些特別的事件記錄只有數(shù)十條。因此將來會考慮對數(shù)據(jù)做更進(jìn)一步的預(yù)處理來減少用戶重復(fù)進(jìn)行預(yù)處理的時(shí)間。

對于使用機(jī)器學(xué)習(xí)的運(yùn)行任務(wù)，本文平臺也給用戶提供了Jupyter接口來使用Python Notebook。圖3顯示了使用Jupyter調(diào)用Spark的Python接口（Pyspark），在YARN集群上運(yùn)行不同的回歸（regression）分析進(jìn)行預(yù)測的例子（代碼只是示例，從用戶代碼提取而來）。

對用戶代碼倉庫中的代碼，僅調(diào)度選手的代碼倉庫的發(fā)布（release）分支中的代碼。這樣做的好處是有效減少了作業(yè)的總數(shù)量，降低了單個(gè)作業(yè)的響應(yīng)時(shí)間。選手的大量調(diào)試代碼先要在線下對樣本數(shù)據(jù)進(jìn)行調(diào)試，通過后才會調(diào)度，這也促使選手在提交代碼之前認(rèn)真思考代碼的目的，因此線上的代碼質(zhì)量比較高（完整性好，bug較少）。

6 結(jié)束語

本文嘗試完全使用開源軟件設(shè)計(jì)并配置了一個(gè)存儲、計(jì)算及代碼并存的數(shù)據(jù)平臺。用戶不僅可以共享數(shù)據(jù)和計(jì)算，還可以追蹤自己的代碼的改變。平臺可以自動審核用戶代碼，并且自動調(diào)度到數(shù)據(jù)平臺進(jìn)行計(jì)算。選取硬件搭建了一個(gè)生產(chǎn)環(huán)境，并且給出了基本的測試。使用這個(gè)平臺服務(wù)于上海開放數(shù)據(jù)創(chuàng)新應(yīng)用大賽（SODA），給出了大賽使用中的經(jīng)驗(yàn)、分析和見解。將來會考慮對平臺進(jìn)行深度優(yōu)化，并且探索如何在各平臺間進(jìn)行數(shù)據(jù)和計(jì)算的共享。

致謝感謝上海市經(jīng)信委和SODA組委會的大力支持。

[1]Thusoo A,Sarma J S,Jain N,et al.Hive—a warehousing solution over a Map-Reduce framework[J].Proceedings of the VLDB Endowment,2009,2(2):1626-1629.

[2]Zaharia M,Chowdhury M,Franklin M J,et al.Spark:cluster computing with working sets[C]//Proceedings of the 2nd USENIX Workshop on Hot Topics in Cloud Computing,Boston,Jun 22,2010.Berkeley:USENIXAssociation,2010:10.

[3]Bhardwaj A P,Deshpande A,Elmore A J,et al.Collaborative data analytics with DataHub[J].Proceedings of the VLDB Endowment,2015,8(12):1916-1919.

[4]Ghemawat S,Gobioff H,Leung S T.The Google file system[C]//Proceedings of the 19th ACM Symposium on Operating Systems Principles,Bolton Landing,Oct 19-22,2003.New York:ACM,2003:29-43.

[5]Chang F,Dean J,Ghemawat S,et al.Bigtable:a distributed storage system for structured data[C]//Proceedings of the 7th Symposium on Operating Systems Design and Implementation,Seattle,Nov 6-8,2006.Berkeley:USENIX Association,2006:205-218.

[6]Dean J,Ghemawat S.MapReduce:simplified data processing on large clusters[C]//Proceedings of the 6th Symposium on Operating System Design and Implementation,San Francisco,Dec 6-8,2004.Berkeley:USENIX Association,2004:137-150.

[7]Neuman B C,Ts′o T.Kerberos:an authentication service for computer networks[J].IEEE Communications Magazine,1994,32(9):33-38.

[8]Zaharia M,Chowdhury M,Das T,et al.Resilient distributed datasets:a fault-tolerant abstraction for in-memory cluster computing[C]//Proceedings of the 9th USENIX Symposium on Networked Systems Design and Implementation,San Jose,Apr 25-27,2012.Berkeley:USENIX Association,2012:15-28.

[9]Armbrust M,Xin R S,Lian Cheng,et al.Spark SQL:relational data processing in Spark[C]//Proceedings of the 2015 International Conference on Management of Data,Melbourne,May 31-Jun 4,2015.New York:ACM,2015:1383-1394.

[10]Bernard J.Running scientific code using IPython and SciPy[J].Linux Journal,2013(228):3.

[11]DeCandia G,Hastorun D,Jampani M,et al.Dynamo:Amazon's highly available key-value store[C]//Proceedings of the 21st Symposium on Operating Systems Principles,Stevenson,Oct 14-17,2007.New York:ACM,2007:205-220.

[12]Wang Lei,Zhan Jianfeng,Luo Chunjie,et al.BigDataBench:a big data benchmark suite from Internet services[C]//Proceedings of the 20th International Symposium on High Performance Computer Architecture,Orlando,Feb 15-19,2014.Washington:IEEE Computer Society,2014:488-499.

[13]Huang Shengsheng,Huang Jie,Dai Jinquan,et al.The Hi-Bench benchmark suite:characterization of the MapReducebased data analysis[C]//Proceedings of the 29th International Conference on Data Engineering Workshops,Long Beach,Mar 1-6,2010.Washington:IEEE Computer Society,2010:41-51.

[14]Melnik S,Gubarev A,Long Jingjing,et al.Dremel:interactive analysis of Web-scale datasets[J].Communications of theACM,2011,54(6):114-123.

[15]Johnson T.Performance measurements of compressed bitmap indices[C]//Proceedings of the 25th International Conference on Very Large Data Bases,Edinburgh Sep 7-10,1999.San Francisco:Morgan Kaufmann Publishers Inc,1999:278-289.