第5章分布式處理
　　5.1CPU多核和POSIX Thread
　　為了提高任務(wù)的計(jì)算處理能力，下面分別從硬件和軟件層面研究新的計(jì)算處理能力。
　　在硬件設(shè)備上，CPU技術(shù)不斷發(fā)展，出現(xiàn)了SMP(對(duì)稱(chēng)多處理器)和NUMA(非一致性?xún)?nèi)存訪(fǎng)問(wèn))兩種高速處理的CPU結(jié)構(gòu)。處理器性能的提升給大量的任務(wù)處理提供了很大的發(fā)展空間。圖5.1是SMP和NUMA結(jié)構(gòu)的CPU，CPU核數(shù)的增加帶來(lái)了計(jì)算能力的提高，但是也隨之帶來(lái)了大量的問(wèn)題需要解決，比如CPU緩存一致性問(wèn)題、NUMA內(nèi)存分配策略等，目前已經(jīng)有比較不錯(cuò)的解決方案。
　　圖5.1SMP和NUMA架構(gòu)CPU
　　在軟件層面出現(xiàn)了多進(jìn)程和多線(xiàn)程編程。進(jìn)程是內(nèi)存資源管理單元，線(xiàn)程是任務(wù)調(diào)度單元。圖5.2是進(jìn)程和線(xiàn)程之間的區(qū)別。
　　圖5.2進(jìn)程與線(xiàn)程
　　總的來(lái)說(shuō)，線(xiàn)程所占用的資源更少，運(yùn)行一個(gè)線(xiàn)程所需要的資源包括寄存器、棧、程序計(jì)數(shù)器等。早期不同廠(chǎng)商提供了不同的多線(xiàn)程編寫(xiě)庫(kù)，這些線(xiàn)程庫(kù)差異巨大，為了統(tǒng)一多種不同的多線(xiàn)程庫(kù)，共同制定了POSIX Thread多線(xiàn)程編程標(biāo)準(zhǔn)，以充分利用多個(gè)不同的線(xiàn)程庫(kù)。組成POSIX Thread的API分成以下4個(gè)大類(lèi)：
　�。�1） 線(xiàn)程管理。線(xiàn)程管理主要負(fù)責(zé)線(xiàn)程的create、detach、join等，也包括線(xiàn)程屬性的查詢(xún)和設(shè)置。
　　（2） mutexes。處理同步的例程(routine)稱(chēng)為mutex，mutex提供了create、destroy、lock和unlock等函數(shù)。
　　（3） 條件變量。條件變量主要用于多個(gè)線(xiàn)程之間的通信和協(xié)調(diào)。
　�。�4） 同步。同步用于管理讀/寫(xiě)鎖以及barriers。
　　5.2MPI并行計(jì)算框架
　　MPI(Message Passing Interface)是一個(gè)標(biāo)準(zhǔn)且可移植的消息傳遞系統(tǒng)，服務(wù)于大規(guī)模的并行計(jì)算。MPI標(biāo)準(zhǔn)定義了采用C、C++、Fortran語(yǔ)言編寫(xiě)程序的函數(shù)語(yǔ)法和語(yǔ)義。目前有很多經(jīng)過(guò)良好測(cè)試和高效率的關(guān)于MPI的實(shí)現(xiàn)，廣泛采用的實(shí)現(xiàn)有MPICH。下面以MPICH為例展開(kāi)對(duì)MPI的講解。
　　MPICH是一個(gè)高性能且可以廣泛移植的MPI實(shí)現(xiàn)。圖5.3為MPICH的架構(gòu)圖。
　　圖5.3MPICH架構(gòu)
　　如圖5.3所示，應(yīng)用程序通過(guò)MPI結(jié)構(gòu)連接到MPICH接口層，圖中的ROMIO是MPI.IO的具體實(shí)現(xiàn)版本，對(duì)應(yīng)MPI標(biāo)準(zhǔn)中的高性能實(shí)現(xiàn)。MPICH包括ADI3、CH3 Device、CH3 Interface、Nemesis、Nemesis NetMod Interface。
　　（1) ADI3。ADI是抽象設(shè)備接口(abstract device interface)，MPICH通過(guò)ADI3接口層隔離底層的具體設(shè)備。
　　（2) CH3 Device。CH3 Device是ADI3的一個(gè)具體實(shí)現(xiàn)，使用了相對(duì)少數(shù)目的函數(shù)功能。在CH3 Device實(shí)現(xiàn)了多個(gè)通信channel，channel提供了兩個(gè)MPI進(jìn)程之間傳遞數(shù)據(jù)的途徑以及進(jìn)程通信。當(dāng)前包括兩個(gè)channel，即Nemesis和Sock，其中Sock是一個(gè)基于UNIX Socket的channel，而Nemesis支持多種方法，不僅僅局限于Socket通信。
　　（3) CH3 Interface。CH3 Inferface用于定義訪(fǎng)問(wèn)Nemesis的接口規(guī)范。
　�。�4） Nemesis。Nemesis允許兩個(gè)MPI進(jìn)程之間的網(wǎng)絡(luò)通信采取多種方法，包括TCP、InfiniBand等。
　　5.3Hadoop MapReduce
　　Hadoop是一個(gè)由Apache基金會(huì)開(kāi)發(fā)的分布式系統(tǒng)基礎(chǔ)架構(gòu)。Hadoop框架*核心的設(shè)計(jì)就是HDFS和MapReduce，HDFS為海量的數(shù)據(jù)提供了存儲(chǔ)，而MapReduce為海量的數(shù)據(jù)提供了計(jì)算。
　　HDFS(Hadoop Distributed File System)有高容錯(cuò)性的特點(diǎn)，并且設(shè)計(jì)用來(lái)部署在低廉的硬件上； 而且它提供高吞吐量來(lái)訪(fǎng)問(wèn)應(yīng)用程序的數(shù)據(jù)，適合有著超大數(shù)據(jù)集的應(yīng)用程序。HDFS放寬了POSIX的要求，可以用流的形式訪(fǎng)問(wèn)文件系統(tǒng)中的數(shù)據(jù)。
　　MapReduce是Google公司提出的一個(gè)軟件框架，用于大規(guī)模數(shù)據(jù)集(大于1TB)的并行運(yùn)算。“Map”和“Reduce”概念以及它們的主要思想都是從函數(shù)式編程語(yǔ)言借來(lái)的，還有從矢量編程語(yǔ)言借來(lái)的特性。
　　當(dāng)前的軟件實(shí)現(xiàn)是指定一個(gè)Map函數(shù)，用來(lái)把一組鍵值對(duì)映射成一組新的鍵值對(duì)，指定并發(fā)的Reduce函數(shù)，用來(lái)保證所有映射的鍵值對(duì)中的每一個(gè)共享相同的鍵組。
　　處理流程如下：
　　（1） MapReduce框架將應(yīng)用的輸入數(shù)據(jù)切分成M個(gè)模塊，典型的數(shù)據(jù)塊大小為64MB。
　�。�2） 具有全局*的主控Master以及若干個(gè)Worker，Master負(fù)責(zé)為Worker分配具體的Map或Reduce任務(wù)并做全局管理。
　　（3） Map任務(wù)的Worker讀取對(duì)應(yīng)的數(shù)據(jù)塊內(nèi)容，從數(shù)據(jù)塊中解析Key/Value記錄數(shù)據(jù)并將其傳給用戶(hù)自定義的Map函數(shù)，Map函數(shù)輸出的中間結(jié)果Key/Value數(shù)據(jù)在內(nèi)存中緩存。
　�。�4） 緩存的Map函數(shù)產(chǎn)生的中間結(jié)果周期性地寫(xiě)入磁盤(pán)，每個(gè)Map函數(shù)中間結(jié)果在寫(xiě)入磁盤(pán)前被分割函數(shù)切割成R份，R是Reduce的個(gè)數(shù)。一般用Key對(duì)R進(jìn)行哈希取模。Map函數(shù)完成對(duì)應(yīng)數(shù)據(jù)塊處理后將R個(gè)臨時(shí)文件位置通知Master，Master再轉(zhuǎn)交給Reduce任務(wù)的Worker。
　　（5） Reduce任務(wù)Worker接到通知時(shí)將Map產(chǎn)生的M份數(shù)據(jù)文件pull到本地(當(dāng)且僅當(dāng)所有Map函數(shù)完成時(shí)Reduce函數(shù)才能執(zhí)行)。Reduce任務(wù)根據(jù)中間數(shù)據(jù)的Key對(duì)記錄進(jìn)行排序，相同Key的記錄聚合在一起。
　　（6） 所有Map、Reduce任務(wù)完成，Master喚醒用戶(hù)應(yīng)用程序。
　　5.4Spark
　　Spark是UC Berkeley AMP Lab所開(kāi)源的類(lèi)Hadoop MapReduce的通用的并行計(jì)算框架，Spark基于Map.Reduce算法實(shí)現(xiàn)的分布式計(jì)算，擁有Hadoop MapReduce所具有的優(yōu)點(diǎn)； 不同于MapReduce的是中間輸出和結(jié)果可以保存在內(nèi)存中，從而不再需要讀/寫(xiě)HDFS，因此Spark能更好地適用于數(shù)據(jù)挖掘與機(jī)器學(xué)習(xí)等需要迭代的Map.Reduce的算法。
　　Spark*主要的結(jié)構(gòu)是RDD(Resilient Distributed Datasets),它表示已被分區(qū)、不可變的并能夠被并行操作的數(shù)據(jù)集合，不同的數(shù)據(jù)集格式對(duì)應(yīng)不同的RDD實(shí)現(xiàn)。RDD必須是可序列化的。RDD可以緩存到內(nèi)存中，每次對(duì)RDD數(shù)據(jù)集操作之后的結(jié)果都可以存放到內(nèi)存中，下一個(gè)操作可以直接從內(nèi)存中輸入，省去了MapReduce大量的磁盤(pán)I/O操作。這很適合迭代運(yùn)算比較常見(jiàn)的機(jī)器學(xué)習(xí)算法、交互式數(shù)據(jù)挖掘。
　　與Hadoop類(lèi)似，Spark支持單節(jié)點(diǎn)集群或多節(jié)點(diǎn)集群。對(duì)于多節(jié)點(diǎn)操作，Spark可以采用自己的資源管理器，也可以采用Mesos集群管理器來(lái)管理資源。Mesos為分布式應(yīng)用程序的資源共享和隔離提供了一個(gè)有效平臺(tái)(參見(jiàn)圖5.4)。該設(shè)置允許Spark與Hadoop共存于節(jié)點(diǎn)的一個(gè)共享池中。
　　圖5.4Mesos集群管理器
　　5.5數(shù)據(jù)處理技術(shù)的發(fā)展
　　數(shù)據(jù)處理從早期的共享分時(shí)單CPU操作系統(tǒng)處理到多核并發(fā)處理，每臺(tái)計(jì)算機(jī)設(shè)備的處理能力在不斷增強(qiáng)，處理的任務(wù)復(fù)雜度在不斷增加，任務(wù)的處理時(shí)間在不斷減少。
　　然而，隨著大數(shù)據(jù)技術(shù)的不斷發(fā)展，一臺(tái)計(jì)算設(shè)備無(wú)法勝任目前大數(shù)據(jù)計(jì)算的龐大的計(jì)算工作。為了解決單臺(tái)計(jì)算機(jī)無(wú)法處理大規(guī)模數(shù)據(jù)計(jì)算的問(wèn)題，連接多臺(tái)計(jì)算機(jī)設(shè)備整合成一個(gè)統(tǒng)一的計(jì)算系統(tǒng)，對(duì)外提供計(jì)算服務(wù)。早期Google公司的分布式計(jì)算框架MapReduce采用的思想就是連接多臺(tái)廉價(jià)的計(jì)算設(shè)備，以此來(lái)提供進(jìn)行大規(guī)模計(jì)算任務(wù)的能力。但是MapReduce是建立在磁盤(pán)之上的并行計(jì)算框架，由于機(jī)械磁盤(pán)本身的局限性，MapReduce仍然有很大的計(jì)算延遲。Spark提出了把計(jì)算結(jié)果存放在內(nèi)存中,利用內(nèi)存作為存儲(chǔ)介質(zhì)的方法極大地縮短了系統(tǒng)的響應(yīng)時(shí)間,降低了計(jì)算任務(wù)返回結(jié)果的延遲。為了滿(mǎn)足大規(guī)模機(jī)器學(xué)習(xí)計(jì)算任務(wù)的需求,也設(shè)計(jì)了大量的分布式機(jī)器學(xué)習(xí)框架來(lái)訓(xùn)練機(jī)器模型參數(shù),比如Parameter Server;針對(duì)圖計(jì)算場(chǎng)合,Google公司設(shè)計(jì)實(shí)現(xiàn)了Pregel圖計(jì)算框架，用于處理*短路徑、Dijstra等經(jīng)典圖計(jì)算任務(wù)； 為了滿(mǎn)足實(shí)時(shí)計(jì)算任務(wù)需求，設(shè)計(jì)實(shí)現(xiàn)了流計(jì)算框架，比如Spark Streaming、Storm、Flink等實(shí)時(shí)計(jì)算框架。
　　總之，目前處理技術(shù)在往大規(guī)模、低延遲方向發(fā)展，內(nèi)存空間的擴(kuò)大以及內(nèi)存存儲(chǔ)成本的降低給大規(guī)模數(shù)據(jù)處理提供了極好的發(fā)展契機(jī)。
　　5.6習(xí)題
　　1. 簡(jiǎn)述CPU技術(shù)的發(fā)展趨勢(shì)。
　　2. 簡(jiǎn)述MPICH并行計(jì)算框架。
　　3. 簡(jiǎn)述MapReduce的原理。
　　……