Hadoop簡明教程 課件 第3、4章 分布式文件系統(tǒng)HDFS、分布式計(jì)算框架MapReduce

第3章分布式文件系統(tǒng)HDFS3.1HDFS的架構(gòu)和原理嵌入式系統(tǒng)基本概念相對(duì)于傳統(tǒng)的本地文件系統(tǒng)，分布式文件系統(tǒng)是一種通過網(wǎng)絡(luò)實(shí)現(xiàn)文件在多臺(tái)計(jì)算機(jī)上進(jìn)行分布式存儲(chǔ)的文件系統(tǒng)。HDFS是一個(gè)高度容錯(cuò)性的系統(tǒng)，適合部署在廉價(jià)的機(jī)器上。HDFS能提供高吞吐量的數(shù)據(jù)訪問，非常適合大規(guī)模數(shù)據(jù)集上的應(yīng)用。HDFS是一個(gè)主從結(jié)構(gòu)，一個(gè)HDFS集群主要由一個(gè)NameNode和一些DataNode組成。NameNode在HDFS內(nèi)部提供元數(shù)據(jù)服務(wù)；DataNode為HDFS提供數(shù)據(jù)的存儲(chǔ)和讀取。3.1.1計(jì)算機(jī)集群結(jié)構(gòu)嵌入式系統(tǒng)基本概念

圖3-1計(jì)算機(jī)集群的基本架構(gòu)3.1.2HDFS的假設(shè)前提和設(shè)計(jì)目標(biāo)嵌入式系統(tǒng)基本概念1.硬件錯(cuò)誤的處理2.流式數(shù)據(jù)訪問3.大規(guī)模數(shù)據(jù)集4.簡單的一致性模型5.移動(dòng)計(jì)算比移動(dòng)數(shù)據(jù)更劃算6.異構(gòu)軟硬件平臺(tái)間的可移植性3.1.3HDFS的相關(guān)概念嵌入式系統(tǒng)基本概念

1.塊HDFS也同樣采用了塊的概念，默認(rèn)的一個(gè)數(shù)據(jù)塊大小為128MB。存放在HDFS上文件會(huì)被拆分成多個(gè)數(shù)據(jù)塊，每個(gè)塊作為獨(dú)立的單位進(jìn)行存儲(chǔ)。不同于普通文件系統(tǒng)，HDFS中如果一個(gè)文件小于一個(gè)數(shù)據(jù)塊的大小，并不占用整個(gè)數(shù)據(jù)塊存儲(chǔ)空間。除磁盤尋道開銷以外，HDFS還有數(shù)據(jù)塊的定位開銷，當(dāng)客戶端訪問一個(gè)HDFS文件時(shí)，首先從NameNode獲取該文件的各數(shù)據(jù)塊位置列表，然后根據(jù)位置列表獲取存儲(chǔ)各數(shù)據(jù)塊的DataNode位置，最后，DataNode在本地文件系統(tǒng)中找出對(duì)應(yīng)的文件，并將數(shù)據(jù)返回給客戶端。3.1.3HDFS的相關(guān)概念嵌入式系統(tǒng)基本概念

2.元數(shù)據(jù)元數(shù)據(jù)是描述數(shù)據(jù)的數(shù)據(jù)，主要是描述數(shù)據(jù)屬性的信息，用來支持如指示存儲(chǔ)位置、歷史數(shù)據(jù)、資源查找、文件記錄等功能。在HDFS中，元數(shù)據(jù)主要有三類信息：第一類是目錄和文件自身的屬性信息，如目錄名、父目錄信息、文件名、文件大小、創(chuàng)建時(shí)間和修改時(shí)間等；第二類記錄與文件存儲(chǔ)相關(guān)的信息,如文件分塊情況、副本個(gè)數(shù)、每個(gè)副本所在的DataNode信息等；第三類用來記錄HDFS中所有DataNode的信息，用于管理集群的DataNode。3.1.3HDFS的相關(guān)概念嵌入式系統(tǒng)基本概念3.NameNode

NameNode（名稱節(jié)點(diǎn)）是整個(gè)文件系統(tǒng)的管理節(jié)點(diǎn)，負(fù)責(zé)維護(hù)文件系統(tǒng)的命名空間，任何對(duì)文件系統(tǒng)命名空間或?qū)傩缘男薷亩紝⒈籒ameNode記錄下來。應(yīng)用程序可以設(shè)置HDFS保存的文件的副本數(shù)目。文件副本的數(shù)目稱為文件的副本系數(shù)，這個(gè)信息也是由NameNode保存的。NameNode還記錄了每個(gè)文件中各個(gè)塊所在的數(shù)據(jù)節(jié)點(diǎn)的位置信息。3.1.3HDFS的相關(guān)概念嵌入式系統(tǒng)基本概念

圖3-2NameNode的核心數(shù)據(jù)結(jié)構(gòu)3.1.3HDFS的相關(guān)概念嵌入式系統(tǒng)基本概念

4.DataNode

DataNode（數(shù)據(jù)節(jié)點(diǎn)）是真正存儲(chǔ)數(shù)據(jù)的地方。HDFS數(shù)據(jù)存儲(chǔ)在DataNode上，數(shù)據(jù)塊的創(chuàng)建、復(fù)制和刪除都在DataNode上執(zhí)行。DataNode會(huì)根據(jù)客戶端或者是NameNode的調(diào)度來進(jìn)行數(shù)據(jù)的存儲(chǔ)和檢索，并且向NameNode定期發(fā)送自己所存儲(chǔ)塊的列表。它將HDFS數(shù)據(jù)以文件的形式存儲(chǔ)在本地的Linux文件系統(tǒng)中。3.1.3HDFS的相關(guān)概念嵌入式系統(tǒng)基本概念

5.SecondaryNameNode

在NameNode運(yùn)行期間，HDFS的所有更新操作都是直接寫到EditLog中，隨著時(shí)間的推移，EditLog會(huì)變得越來越大。SecondaryNameNode（第二名稱節(jié)點(diǎn)）負(fù)責(zé)定時(shí)從NameNode上獲取FsImage和EditLog，同時(shí)請(qǐng)求NameNode停止使用EditLog文件，暫時(shí)將新的寫操作寫到一個(gè)新的文件EditLog.new上。在SecondaryNameNode上將FsImage、EditLog合并得到新的FsImage，將新的FsImage復(fù)制到NameNode，取代原來的FsImage，同時(shí)用EditLog.new文件去替換EditLog文件，從而減小了EditLog文件的大小。3.1.4HDFS體系結(jié)構(gòu)嵌入式系統(tǒng)基本概念

HDFS采用了主從（Master/Slave）結(jié)構(gòu)模型，如圖3-3所示。一個(gè)HDFS集群是由一個(gè)NameNode和一定數(shù)目的DataNode組成。NameNode是一個(gè)中心服務(wù)器，負(fù)責(zé)管理文件系統(tǒng)的命名空間以及客戶端對(duì)文件的訪問。集群中的DataNode一般是一個(gè)節(jié)點(diǎn)一個(gè)，負(fù)責(zé)管理它所在節(jié)點(diǎn)上的數(shù)據(jù)存儲(chǔ)。3.1.4HDFS體系結(jié)構(gòu)嵌入式系統(tǒng)基本概念

圖3-3HDFS的體系結(jié)構(gòu)3.1.5HDFS存儲(chǔ)原理嵌入式系統(tǒng)基本概念

1.冗余數(shù)據(jù)保存

HDFS被設(shè)計(jì)成能夠在一個(gè)大集群中跨機(jī)器可靠地存儲(chǔ)超大文件。它將每個(gè)文件存儲(chǔ)成一系列的數(shù)據(jù)塊，除了最后一個(gè)，所有數(shù)據(jù)塊的大小都是相同的。為了保證系統(tǒng)的容錯(cuò)性和可用性，HDFS采用了多副本方式對(duì)數(shù)據(jù)進(jìn)行冗余存儲(chǔ)，通常一個(gè)數(shù)據(jù)塊的多個(gè)副本會(huì)被存儲(chǔ)到不同的DataNode上，每個(gè)文件的數(shù)據(jù)塊大小和副本系數(shù)都是可配置的。應(yīng)用程序可以指定某個(gè)文件的副本數(shù)目。3.1.5HDFS存儲(chǔ)原理嵌入式系統(tǒng)基本概念

2.數(shù)據(jù)存取策略副本的存放是HDFS可靠性和性能的關(guān)鍵。優(yōu)化的副本存放策略是HDFS區(qū)分于其他大部分分布式文件系統(tǒng)的重要特性。HDFS采用一種稱為機(jī)架感知的策略來改進(jìn)數(shù)據(jù)的可靠性、可用性和網(wǎng)絡(luò)帶寬的利用率。通過一個(gè)機(jī)架感知的過程，NameNode可以確定每個(gè)DataNode所屬的機(jī)架id。一個(gè)簡單但沒有優(yōu)化的策略就是將副本存放在不同的機(jī)架上。這樣可以有效防止當(dāng)整個(gè)機(jī)架失效時(shí)丟失數(shù)據(jù)，并且允許讀數(shù)據(jù)的時(shí)候充分利用多個(gè)機(jī)架的帶寬。在默認(rèn)情況下，副本系數(shù)是3。為了降低整體的帶寬消耗和讀取延時(shí)，HDFS會(huì)盡量讓讀取程序讀取離它最近的副本。如果在讀取程序的同一個(gè)機(jī)架上有一個(gè)副本，那么就讀取該副本。3.1.5HDFS存儲(chǔ)原理嵌入式系統(tǒng)基本概念

3.流水線復(fù)制當(dāng)客戶端向HDFS文件寫入數(shù)據(jù)的時(shí)候，一開始是寫到本地臨時(shí)文件中。假設(shè)該文件的副本系數(shù)設(shè)置為3，當(dāng)本地臨時(shí)文件累積到一個(gè)數(shù)據(jù)塊的大小時(shí)，客戶端會(huì)從NameNode獲取一個(gè)DataNode列表用于存放副本。然后客戶端開始向第一個(gè)DataNode傳輸數(shù)據(jù)，第一個(gè)DataNode一小部分一小部分(4KB)地接收數(shù)據(jù)，將每一部分寫入本地，并同時(shí)傳輸該部分到列表中第二個(gè)DataNode節(jié)點(diǎn)。第二個(gè)DataNode也是這樣，一小部分一小部分地接收數(shù)據(jù)，寫入本地，并同時(shí)傳給第三個(gè)DataNode。最后，第三個(gè)DataNode接收數(shù)據(jù)并存儲(chǔ)在本地。因此，DataNode能流水線式地從前一個(gè)節(jié)點(diǎn)接收數(shù)據(jù)，并在同時(shí)轉(zhuǎn)發(fā)給下一個(gè)節(jié)點(diǎn)，數(shù)據(jù)以流水線的方式從前一個(gè)DataNode復(fù)制到下一個(gè)DataNode。3.2HDFSShell嵌入式系統(tǒng)基本概念

1.創(chuàng)建目錄hadoopfs-mkdir<path>：在指定位置創(chuàng)建目錄，用“/”表示根目錄。例如：

hadoopfs-mkdir/test#在HDFS根目錄下創(chuàng)建名為test的目錄hadoopfs-mkdir/test/data#在test目錄下創(chuàng)建名為data的目錄3.2HDFSShell嵌入式系統(tǒng)基本概念

2.查看文件列表hadoopfs-ls<path>：查看指定目錄下的文件列表，例如：

hadoopfs-ls/#查看HDFS根目錄下的子目錄和文件

hadoopfs-ls/test#查看HDFS/test下的子目錄和文件參數(shù)-R可用于查看HDFS指定目錄下所有子目錄和文件，R代表遞歸（Recursive）。使用hadoopfs-ls-R/命令，可以一次性列出所有HDFS子目錄和子目錄下的文件。3.2HDFSShell嵌入式系統(tǒng)基本概念

3.從本地文件系統(tǒng)上傳文件到HDFS將文件從本地文件系統(tǒng)復(fù)制到HDFS稱為文件上傳。有兩種命令可以實(shí)現(xiàn)文件上傳，一種是hadoopfs-put，另一種是hadoopfs-copyFromLocal。例如：hadoopfs-put/user/data/a1.txt/test表示將本地文件系統(tǒng)的“/user/data/a1.txt”上傳到HDFS的“/test”下。hadoopfs-put/user/data/a2.txt/user/data/a3.txt/test表示將本地文件系統(tǒng)的“/user/data/a2.txt”和“/user/data/a3.txt”上傳到HDFS的“/test”下。3.2HDFSShell嵌入式系統(tǒng)基本概念

4.查看HDFS文件內(nèi)容可以使用hadoopfs-cat、hadoopfs-text、hadoopfs-tail等不同參數(shù)形式查看HDFS的文件內(nèi)容。當(dāng)然，只有文本文件的內(nèi)容可以看清楚，其他類型的文件顯示的可能是亂碼。例如，查看HDFS/test/a1.txt的內(nèi)容，命令如下：

hadoopfs-cat/test/a1.txt3.2HDFSShell嵌入式系統(tǒng)基本概念

5.移動(dòng)HDFS文件移動(dòng)HDFS文件，指的將文件從源路徑移動(dòng)到目標(biāo)路徑。這個(gè)命令允許有多個(gè)源路徑，此時(shí)目標(biāo)路徑必須是一個(gè)目錄。不允許在不同的文件系統(tǒng)間移動(dòng)文件。使用方法：hadoopfs-mvURI[URI…]<dest>

例如，將HDFS/test/a2.txt移動(dòng)到/test/data下，命令如下：

hadoopfs-mv/test/a2.txt/test/data3.2HDFSShell嵌入式系統(tǒng)基本概念

6.復(fù)制HDFS文件復(fù)制HDFS文件是將文件從源路徑復(fù)制到目標(biāo)路徑。這個(gè)命令允許有多個(gè)源路徑，此時(shí)目標(biāo)路徑必須是一個(gè)目錄。使用方法：hadoopfs-cpURI[URI…]<dest>

例如，將HDFS/test/a3.txt復(fù)制到/test/data下，命令如下：

hadoopfs-cp/test/a3.txt/test/data3.2HDFSShell嵌入式系統(tǒng)基本概念

7.將HDFS上的文件下載到本地將文件從HDFS復(fù)制到本地稱為文件下載。有兩種命令可以實(shí)現(xiàn)文件下載，一種是hadoopfs-get，另一種是hadoopfs-copyToLocal。兩種的用法相同，下面以第一種為例介紹使用方法。例如：

mkdir/user/data/xiaz在本地路徑/user/data下創(chuàng)建目錄xiaz，用于存儲(chǔ)下載文件。

3.2HDFSShell嵌入式系統(tǒng)基本概念

8.刪除HDFS上的文件可用hadoopfs-rm刪除HDFS上的文件。例如：

hadoopfs-rm/test/data/a2.txt/test/data/a3.txt此命令將HDFS上的/test/data/a2.txt和/test/data/a3.txt刪除掉。可用hadoopfs-rm-R刪除HDFS上的目錄。如果沒有-R，則不能刪除目錄。例如：

hadoopfs-rm-R/test/data/math1此命令將HDFS上/test/data/路徑下的目錄math1及math1下的文件刪除掉。3.3HDFSJavaAPI嵌入式系統(tǒng)基本概念Hadoop使用Java語言編寫，提供了豐富的JavaAPI供開發(fā)人員調(diào)用。HDFSShell本質(zhì)上就是對(duì)JavaAPI的應(yīng)用，凡是使用Shell命令可以完成的功能，都可以使用JavaAPI來實(shí)現(xiàn)。3.3.2HDFSJavaAPI簡介嵌入式系統(tǒng)基本概念

1.Configuration類Configuration類將HDFS的客戶端或服務(wù)器端的配置信息傳遞給FileSystem。具體來講，它會(huì)加載etc/Hadoop/core-site.xml文件中的配置信息。Configuration類的使用方法如下：Configurationconf=newConfiguration();

conf.set("fs.defaultFS","hdfs://30:9000");3.3.2HDFSJavaAPI簡介嵌入式系統(tǒng)基本概念

2.FileSystem類

FileSystem是一個(gè)通用文件系統(tǒng)的抽象基類，可以被分布式文件系統(tǒng)繼承，所有可能使用Hadoop文件系統(tǒng)的程序，都要使用這個(gè)類。Hadoop為FileSystem這個(gè)抽象類提供了多種具體實(shí)現(xiàn)，DistributedFileSystem就是FileSystem在HDFS文件系統(tǒng)中的具體實(shí)現(xiàn)。該類是個(gè)抽象類，只能通過類的get方法來得到具體對(duì)象。3.3.2HDFSJavaAPI簡介嵌入式系統(tǒng)基本概念

FileSystem類的常見方法如下：（1）publicstaticFileSystemget(Configurationconf)throwsIOException：根據(jù)配置信息返回文件系統(tǒng)實(shí)例。（2）publicstaticFileSystemget(URIuri,Configurationconf)throwsIOException：根據(jù)URI方案和配置信息返回文件系統(tǒng)實(shí)例。3.3.2HDFSJavaAPI簡介嵌入式系統(tǒng)基本概念（3）publicFSDataOutputStreamcreate(Pathf,Booleanoverwrite)throwsIOException：創(chuàng)建文件，返回一個(gè)輸出流FSDataOutputStream對(duì)象，其中f用于指定文件路徑，overwrite用于指定是否覆蓋現(xiàn)有文件。注意：該方法有多種重載形式。（4）publicabstractFSDataInputStreamopen(Pathf,int

bufferSize)throwsIOException：打開文件，返回一個(gè)輸入流FSDataInputStream對(duì)象。其中f用于指定文件路徑，bufferSize用于指定緩沖區(qū)的大小。注意：該方法有多種重載形式。3.3.2HDFSJavaAPI簡介嵌入式系統(tǒng)基本概念（5）publicboolean

mkdirs(Pathf)throwsIOException：創(chuàng)建目錄和子目錄，其中f是完整的目錄路徑。創(chuàng)建成功后返回true，否則返回false。（6）publicabstractbooleandelete(Pathf,booleanrecursive)throwsIOException：刪除文件或目錄，如果要同時(shí)刪除子目錄（非空目錄），則需要設(shè)置參數(shù)recursive為true。如果刪除失?。ū热缥募淮嬖冢﹦t會(huì)拋出I/O異常。3.3.2HDFSJavaAPI簡介嵌入式系統(tǒng)基本概念（7）publicvoidcopyFromLocalFile(Pathsrc,Pathdst)throwsIOException：將本地文件系統(tǒng)的文件復(fù)制到HDFS，其中src為本地文件系統(tǒng)的文件路徑，dst為HDFS上的目錄路徑。（8）publicvoidcopyToLocalFile(Pathsrc,Path

dst)throwsIOException：將HDFS上的文件復(fù)制到本地文件系統(tǒng)，其中src為HDFS的文件路徑，dst為本地文件系統(tǒng)的目錄路徑。3.3.2HDFSJavaAPI簡介嵌入式系統(tǒng)基本概念（9）publicbooleanexists(Pathf)throwsIOException：可查看指定的HDFS文件是否存在，其中f用于指定文件路徑。（10）publicabstractbooleanrename(Pathsrc,Path

dst)throwsIOException：可為指定的HDFS目錄/文件重命名，src表示要重命名的目錄/文件路徑，dst表示重命名后的目錄/文件新路徑，重命名成功后返回true，否則返回false。（11）publicabstractFileStatus

getFileStatus(Pathf)throwsIOException：獲取文件系統(tǒng)的目錄和文件的元數(shù)據(jù)信息。3.3.2HDFSJavaAPI簡介嵌入式系統(tǒng)基本概念

3.FSDataInputStream類

FileSystem對(duì)象調(diào)用open()方法返回的是FSDataInputStream的對(duì)象，而不是標(biāo)準(zhǔn)的java.io類對(duì)象，F(xiàn)SDataInputStream類繼承了java.io.DataInputStream類。這個(gè)類重載了多種read方法，用于讀取多種類型的數(shù)據(jù)，下面是一種常用的read方法。publicintread(ByteBuffer

buf)throwsIOException：輸入流對(duì)象調(diào)用該方法從源中試圖讀取buf長度個(gè)字節(jié)數(shù)據(jù)，并將它們存儲(chǔ)到緩沖區(qū)buf中，返回值為實(shí)際讀取的字節(jié)數(shù)。3.3.2HDFSJavaAPI簡介嵌入式系統(tǒng)基本概念

4.FSDataOutputStream類

FileSystem對(duì)象調(diào)用create()方法返回的是FSDataOutputStream的對(duì)象，F(xiàn)SDataOutputStream類繼承了java.io.DataOutputStream類。該類繼承write方法。publicvoidwrite(byte[]b)throwsIOException：輸出流對(duì)象調(diào)用該方法向輸出流寫入一個(gè)字節(jié)數(shù)組。3.3.2HDFSJavaAPI簡介嵌入式系統(tǒng)基本概念

5.FileStatus類

FileStatus類封裝了HDFS中文件和目錄的元數(shù)據(jù)信息，包括文件大小、存放路徑、塊大小、備份數(shù)、所有者、修改時(shí)間以及權(quán)限等信息。3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

【例3-1】在HDFS上創(chuàng)建文本文件，并向該文件中寫入文本信息，然后將該文件內(nèi)容讀出。3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FSDataInputStream;importorg.apache.hadoop.fs.FSDataOutputStream;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念publicclassWriteandRead{ publicstaticvoidmain(String[]args){try{/**獲取Hadoop配置信息*/Configurationconf=newConfiguration();

conf.set("fs.defaultFS","hdfs://30:9000");/**創(chuàng)建文件系統(tǒng)實(shí)例對(duì)象fs*/

FileSystemfs=FileSystem.get(conf);/**創(chuàng)建Path路徑實(shí)例*/Pathfile=newPath("hdfs://30:9000/test/t5.txt");3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念/**創(chuàng)建輸出流*/FSDataOutputStream

putOut=fs.create(file);StringxString="HelloHadoop!";/**設(shè)置輸出字節(jié)數(shù)組，并寫入輸出流*/

putOut.write(xString.getBytes());

putOut.close();//關(guān)閉輸出流

FSDataInputStream

getIn=fs.open(file);//創(chuàng)建輸入流byteb1[]=newbyte[100];//創(chuàng)建緩存數(shù)組inta=getIn.read(b1);//將數(shù)據(jù)讀入緩存數(shù)組3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

/**根據(jù)實(shí)際讀入字節(jié)數(shù)生成字符串*/Stringcontent=newString(b1,0,a);

System.out.println(content);

getIn.close();//關(guān)閉輸入流

fs.close();}catch(Exceptione){

e.printStackTrace();}}}3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

圖3-33【例3-1】程序的執(zhí)行結(jié)果3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

【例3-2】使用HDFSJavaAPI操作HDFS上的文件和目錄。主要實(shí)現(xiàn)了如何獲取FileSystem實(shí)例、上傳文件、下載文件、重命名文件、創(chuàng)建和刪除目錄等功能。3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念importjava.io.IOException;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;publicclassHdfsApi{ /**定義一個(gè)fs變量*/

FileSystemfs=null;3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

/**創(chuàng)建具體文件系統(tǒng)對(duì)象*/publicvoidgetfileSystem()throwsIOException{Configurationconf=newConfiguration();

conf.set("fs.defaultFS","hdfs://30:9000");fs=FileSystem.get(conf);}3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

/**本地文件上傳到HDFS，其中src為本地文件系統(tǒng)的文件路徑，dst為HDFS上的目錄路徑*/publicvoidtestAddFileToHdfs(Stringsrc,String

dst)throwsIOException{ PathsrcPath=newPath(src); PathdstPath=newPath(dst);

fs.copyFromLocalFile(srcPath,dstPath); }3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

/**下載HDFS文件到本地，其中src為HDFS的文件路徑，dst為本地文件系統(tǒng)的目錄路徑*/publicvoidtestAddFileToLocal(Stringsrc,String

dst)throwsIOException{ PathsrcPath=newPath(src); PathdstPath=newPath(dst);

fs.copyToLocalFile(srcPath,dstPath); }3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

/**創(chuàng)建HDFS目錄，mk為HDFS上的目錄路徑*/publicvoidtestMakeDir(Stringmk)throwsIOException{

boolean

isSuccess=fs.mkdirs(newPath(mk));if(isSuccess==true)

System.out.println("創(chuàng)建成功"); else

System.out.println("創(chuàng)建失敗");}3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

/**刪除HDFS目錄/文件，del為指定要?jiǎng)h除的目錄/文件，recursive控制是否遞歸刪除*/publicvoidtestDel(Stringdel,booleanrecursive)throwsIOException{

booleanboo=fs.delete(newPath(del),recursive); if(boo==true)

System.out.println("刪除成功"); else

System.out.println("刪除失敗");}3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

/**為指定的HDFS目錄/文件重命名，oldn表示要重命名的目錄/文件路徑，newn表示重命名后的目錄/文件新路徑*/publicvoidrename(Stringoldn,String

newn)throwsIOException{ PatholdPath=newPath(oldn); PathnewPath=newPath(newn);

boolean

isExists=fs.exists(oldPath); if(isExists==true){

boolean

boorename=fs.rename(oldPath,newPath);

System.out.println(boorename?"修改成功！":"修改失??！"); } else

System.out.println(oldn+"文件不存在");}3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

publicstaticvoidmain(Stringargs[]){

HdfsApi

hApi=newHdfsApi(); try{hApi.getfileSystem();

hApi.testAddFileToHdfs("/user/data/a2.txt","/test");

hApi.testAddFileToLocal("/test/a3.txt","/user/data/xiazai");

hApi.testMakeDir("/test/dir1");

hApi.testDel("/test/temp",true);

hApi.rename("/test/zhangsan.txt","/test/lisi.txt");

hApi.fs.close(); }catch(Exceptione){

e.printStackTrace();}}}3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念使用HDFSJavaAPI不但可以實(shí)現(xiàn)對(duì)文件和目錄的各種操作，而且可以讀取文件和目錄的元數(shù)據(jù)信息，下面的例子程序演示了通過FileStatus類讀取各種元數(shù)據(jù)信息?！纠?-3】HDFSAPIFileStatus類的應(yīng)用3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FileStatus;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;publicclassFileStatusTest{ publicstaticvoidmain(String[]args){3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

try{ Configurationconf=newConfiguration();

conf.set("fs.defaultFS","hdfs://30:9000");//獲取Hadoop配置信息

FileSystemfs=FileSystem.get(conf);//創(chuàng)建文件系統(tǒng)實(shí)例對(duì)象fsPathfile1=newPath("/test/t5.txt");//設(shè)置一個(gè)文件路徑

FileStatus

fileStatus=fs.getFileStatus(file1);//獲取文件元數(shù)據(jù)信息

System.out.println("文件的絕對(duì)路徑："+fileStatus.getPath());

System.out.println("塊的大?。?+fileStatus.getBlockSize());

System.out.println("文件的所有者："+fileStatus.getOwner()+":"+fileStatus.getGroup());3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

System.out.println("文件的長度："+fileStatus.getLen());

System.out.println("文件權(quán)限："+fileStatus.getPermission());

System.out.println("修改時(shí)間："+fileStatus.getModificationTime());Pathfile2=newPath("/test");//設(shè)置一個(gè)目錄路徑

FileStatus[]st1=fs.listStatus(file2);for(inti=0;i<st1.length;i++){

System.out.println("文件及所在位置："+st1[i].getPath().toString());}}catch(Exceptione){

e.printStackTrace();}}}3.3.3HDFSJavaAPI編程嵌入式系統(tǒng)基本概念

圖3-34FileStatus類的應(yīng)用程序運(yùn)行結(jié)果第4章分布式計(jì)算框架MapReduce4.1認(rèn)識(shí)MapReduce嵌入式系統(tǒng)基本概念為了降低軟件開發(fā)人員的編程難度，MapReduce框架隱藏了很多內(nèi)部功能的實(shí)現(xiàn)細(xì)節(jié)，實(shí)現(xiàn)了自動(dòng)并行處理。軟件開發(fā)人員可以在不會(huì)分布式并行編程的情況下，將自己的程序運(yùn)行在分布式系統(tǒng)上，完成大數(shù)據(jù)集的計(jì)算。4.1.1分布式并行編程嵌入式系統(tǒng)基本概念單個(gè)計(jì)算機(jī)的CPU、內(nèi)存和硬盤等資源是有限的，單個(gè)計(jì)算機(jī)在處理海量數(shù)據(jù)時(shí)，無法高效地完成大量運(yùn)算。一種有效的解決方案是借助于分布式并行編程來提高程序性能。分布式程序運(yùn)行在大規(guī)模計(jì)算機(jī)集群上，可以并行執(zhí)行大規(guī)模數(shù)據(jù)處理任務(wù)，從而獲得海量的計(jì)算能力。同時(shí)通過向集群中增加新的計(jì)算機(jī)，可以很容易地?cái)U(kuò)充集群的計(jì)算能力。MapReduce最早是由Google提出的一種面向大規(guī)模數(shù)據(jù)處理的分布式并行計(jì)算模型，HadoopMapReduce是它的開源實(shí)現(xiàn)。4.1.2MapReduce簡介嵌入式系統(tǒng)基本概念用MapReduce來處理的數(shù)據(jù)集必須具備這樣的特點(diǎn)：待處理的數(shù)據(jù)集可以分解成許多小的數(shù)據(jù)集，而且每一個(gè)小數(shù)據(jù)集都可以完全獨(dú)立地進(jìn)行并行處理。MapReduce采用“分而治之”的核心思想，即將一個(gè)大數(shù)據(jù)集通過一定的數(shù)據(jù)劃分方法，分成多個(gè)較小的數(shù)據(jù)集，各小的數(shù)據(jù)集之間不存在依賴關(guān)系，將這些小的數(shù)據(jù)集分配給不同的節(jié)點(diǎn)去處理，最后將處理的結(jié)果進(jìn)行匯總，從而形成最終的結(jié)果。4.1.2MapReduce簡介嵌入式系統(tǒng)基本概念

MapReduce設(shè)計(jì)的一個(gè)理念就是“計(jì)算向數(shù)據(jù)靠攏”，因?yàn)樵诖笠?guī)模數(shù)據(jù)環(huán)境下，移動(dòng)數(shù)據(jù)需要大量的網(wǎng)絡(luò)傳輸開銷。只要有可能，MapReduce就會(huì)將Map程序分配到數(shù)據(jù)所在的節(jié)點(diǎn)上運(yùn)行，從而減少節(jié)點(diǎn)之間的數(shù)據(jù)傳輸開銷。最簡單的MapReduce應(yīng)用程序至少包含3個(gè)部分：一個(gè)map函數(shù)、一個(gè)reduce函數(shù)和一個(gè)main函數(shù)。在運(yùn)行一個(gè)MapReduce程序時(shí)，整個(gè)處理過程被分為Map階段和Reduce階段，每個(gè)階段都是用鍵值對(duì)（key/value）作為輸入和輸出。main函數(shù)是MapReduce應(yīng)用程序的入口，它將文件輸入/輸出和作業(yè)控制結(jié)合起來。4.1.3MapReduce的運(yùn)行環(huán)境嵌入式系統(tǒng)基本概念

YARN（YetAnotherResourceNegotiator，另一種資源協(xié)調(diào)者）是Hadoop2.0中的資源管理和調(diào)度框架，YARN的目標(biāo)就是實(shí)現(xiàn)“一個(gè)集群多個(gè)框架”，即在一個(gè)集群上部署一個(gè)統(tǒng)一的資源調(diào)度管理框架YARN，在YARN之上不但可以運(yùn)行MapReduce，而且還可以運(yùn)行Spark、Storm、Tez等計(jì)算框架。YARN的引入為集群在資源統(tǒng)一管理和數(shù)據(jù)共享等方面帶來了很大方便。4.1.3MapReduce的運(yùn)行環(huán)境嵌入式系統(tǒng)基本概念

YARN采用主從架構(gòu)（Master/Slave），包括：ResourceManager、ApplicationMaster和NodeManager三個(gè)核心組件。ResourceManager運(yùn)行在主節(jié)點(diǎn)，負(fù)責(zé)整個(gè)集群的資源管理和分配。每個(gè)應(yīng)用程序擁有一個(gè)ApplicationMaster，ApplicationMaster

管理一個(gè)在YARN內(nèi)運(yùn)行的應(yīng)用程序?qū)嵗?fù)責(zé)申請(qǐng)資源、任務(wù)調(diào)度和任務(wù)監(jiān)控；NodeManager運(yùn)行在從節(jié)點(diǎn)，整個(gè)集群有多個(gè)NodeManager，負(fù)責(zé)單節(jié)點(diǎn)資源的管理和使用。4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念1.ByteWritable：字節(jié)類型。2.IntWritable：整型類型。3.LongWritable：長整型類型。4.BooleanWritable：布爾類型。5.FloatWritable：單精度浮點(diǎn)數(shù)類型。6.DoubleWritable：雙精度浮點(diǎn)數(shù)類型。7.Text：使用UTF8格式存儲(chǔ)的文本類型。8.NullWritable：空對(duì)象，當(dāng)<key,value>中的key或value為空時(shí)使用。4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念【例4-1】常見Hadoop數(shù)據(jù)類型的應(yīng)用importorg.apache.hadoop.io.DoubleWritable;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.Text;4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念publicclassHadoopDataTypeTest{ /*使用hadoop的Text類型*/ publicstaticvoidtestText(){ System.out.println("testText:"); Texttext=newText("Helloworld!"); System.out.println(text.toString()); System.out.println(text.find("r")); System.out.println(text.getLength()); text.set("Hello,hadoop!"); System.out.println(text.toString()); }4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念/*使用hadoop的IntWritable類型*/ publicstaticvoidtestIntWritable(){ System.out.println("testIntWritable:"); IntWritableintWritable=newIntWritable(5); System.out.println("intWritable="+intWritable); inti=intWritable.get()+1; intWritable.set(i); System.out.println("intWritable="+intWritable); }4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念/*使用hadoop的DoubleWritable類型*/ publicstaticvoidtestDoubleWritable(){ System.out.println("testDoubleWritable:"); DoubleWritabledoubleWritable=newDoubleWritable(6.5); System.out.println("doubleWritable="+doubleWritable); doubled=doubleWritable.get()+1.3; doubleWritable.set(d); System.out.println("doubleWritable="+doubleWritable); }4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念publicstaticvoidmain(Stringargs[]){ testText(); testIntWritable(); testDoubleWritable(); }}4.1.4Hadoop內(nèi)置數(shù)據(jù)類型嵌入式系統(tǒng)基本概念圖4-1程序運(yùn)行結(jié)果4.2.1MapReduce工作流程概述嵌入式系統(tǒng)基本概念圖4-2MapReduce的工作流程4.2.2Shuffle過程分析嵌入式系統(tǒng)基本概念圖4-3MapReduce的Shuffle過程4.3MapReduce入門示例：WordCount嵌入式系統(tǒng)基本概念本節(jié)以Hadoop自帶的WordCount程序?yàn)槔?，分析WordCount的設(shè)計(jì)思路、介紹WordCount編程方法和運(yùn)行步驟。4.3.1WordCount程序任務(wù)嵌入式系統(tǒng)基本概念

表4-1WordCount程序任務(wù)4.3.1WordCount程序任務(wù)嵌入式系統(tǒng)基本概念

表4-2WordCount

的輸入和輸出實(shí)例4.3.2準(zhǔn)備被統(tǒng)計(jì)的文件嵌入式系統(tǒng)基本概念創(chuàng)建f1.txt和f2.txt輸入如下命令，創(chuàng)建test2目錄，上傳f1.txt和f2.txt

hadoopfs-mkdir/test2//在HDFS根目錄下創(chuàng)建test2

hadoopfs-putf1.txt/test2//上傳f1.txt到HDFS的/test2下

hadoopfs-putf2.txt/test2//上傳f2.txt到HDFS的/test2下4.3.3WordCount的執(zhí)行過程分析嵌入式系統(tǒng)基本概念圖4-5拆分輸入數(shù)據(jù)4.3.3WordCount的執(zhí)行過程分析嵌入式系統(tǒng)基本概念圖4-6執(zhí)行用戶定義的map()4.3.3WordCount的執(zhí)行過程分析嵌入式系統(tǒng)基本概念圖4-7使用Combine函數(shù)的Map端Shuffle過程4.3.3WordCount的執(zhí)行過程分析嵌入式系統(tǒng)基本概念圖4-8Reduce端的操作4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念importjava.io.IOException;importjava.util.StringTokenizer;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.Text;4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importorg.apache.hadoop.util.GenericOptionsParser;4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念publicclassWordCount{/**自定義的TokenizerMapper類繼承Mapper類。<Object,Text,Text,IntWritable>分別用來指定Map的輸入key值類型、輸入value值類型、輸出key值類型、輸出value值類型*/publicstaticclassTokenizerMapper

extendsMapper<Object,Text,Text,IntWritable>{privatefinalstaticIntWritableone=newIntWritable(1);privateTextword=newText();4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念/**key記錄的數(shù)據(jù)的偏移量，value是每次split提供給程序讀取的一行數(shù)據(jù)*/publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{/**StringTokenizer用于字符串分解，默認(rèn)的分隔符是空格("")、制表符(\t)、換行符(\n)、回車符(\r)*/

StringTokenizer

itr=newStringTokenizer(value.toString());while(itr.hasMoreTokens()){

word.set(itr.nextToken());

context.write(word,one);//以<word,1>的形式輸出

}}}4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念

/**自定義的IntSumReducer類繼承Reducer類。<Text,IntWritable,Text,IntWritable>分別用來指定Reduce的輸入key值類型、輸入value值類型、輸出key值類型、輸出value值類型*/publicstaticclassIntSumReducer

extendsReducer<Text,IntWritable,Text,IntWritable>{privateIntWritableresult=newIntWritable();4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念/**reduce將輸入的key值作為輸出的key值，將key值對(duì)應(yīng)的list各元素值加起來，作為value值輸出*/publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intsum=0;4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念for(IntWritable

val:values){sum+=val.get();//將相同的單詞對(duì)應(yīng)的值加一起

}

result.set(sum);

context.write(key,result);}}4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念/**main()的輸入?yún)?shù)決定了輸入數(shù)據(jù)的文件位置，以及輸出數(shù)據(jù)存儲(chǔ)的位置；創(chuàng)建一個(gè)Configuration對(duì)象時(shí)，會(huì)獲取Hadoop的配置信息；通過Job的對(duì)象可設(shè)置Hadoop程序運(yùn)行時(shí)的環(huán)境變量*/publicstaticvoidmain(String[]args)throwsException{Configurationconf=newConfiguration();String[]otherArgs=newGenericOptionsParser(conf,args).getRemainingArgs();4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念if(otherArgs.length<2){

System.err.println("Usage:wordcount<in>[<in>...]<out>");

System.exit(2);}Jobjob=Job.getInstance(conf,"wordcount");//實(shí)例化job

job.setJarByClass(WordCount.class);//設(shè)置整個(gè)程序的類名

job.setMapperClass(TokenizerMapper.class);//為job設(shè)置Map類

job.setCombinerClass(IntSumReducer.class);//為job設(shè)置Combine類4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念job.setReducerClass(IntSumReducer.class);//為job設(shè)置Reduce類

job.setOutputKeyClass(Text.class);//為job的輸出數(shù)據(jù)設(shè)置Key類job.setOutputValueClass(IntWritable.class);//為job輸出設(shè)置value類/**為job設(shè)置輸入路徑*/for(inti=0;i<otherArgs.length-1;++i){

FileInputFormat.addInputPath(job,newPath(otherArgs[i]));}4.3.4WordCount編程實(shí)踐嵌入式系統(tǒng)基本概念

/**為job設(shè)置輸出路徑*/

FileOutputFormat.setOutputPath(job,newPath(otherArgs[otherArgs.length-1]));

System.exit(job.waitForCompletion(true)?0:1);}}4.4.1求平均值嵌入式系統(tǒng)基本概念求平均值是數(shù)據(jù)分析中的常用操作，下面以求學(xué)生的平均成績?yōu)槔?，介紹使用MapReduce求平均值的方法。【例4-2】現(xiàn)有某班學(xué)生English、Math、Chinese三門課程的成績，每門課程的成績記錄在一個(gè)文本文件中，請(qǐng)編寫MapReduce程序求每位學(xué)生的平均成績。4.4.1求平均值嵌入式系統(tǒng)基本概念importjava.io.IOException;importjava.util.StringTokenizer;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;4.4.1求平均值嵌入式系統(tǒng)基本概念importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importorg.apache.hadoop.util.GenericOptionsParser;4.4.1求平均值嵌入式系統(tǒng)基本概念publicclassWordCount{

publicstaticclassTokenizerMapper

extendsMapper<Object,Text,Text,IntWritable>{/**去掉finalstatic，將one修改為實(shí)例變量*/

privateIntWritableone=newIntWritable(1);privateTextword=newText();publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{4.4.1求平均值嵌入式系統(tǒng)基本概念

/**將輸入的數(shù)據(jù)首先按行進(jìn)行分割*/

StringTokenizer

itr=newStringTokenizer(value.toString(),"\n");/**分別對(duì)每行進(jìn)行處理*/while(itr.hasMoreTokens()){/**將每行按默認(rèn)分隔符進(jìn)行分割*/

StringTokenizerlie=newStringTokenizer(itr.nextToken());/**獲取學(xué)生姓名*/

StringstrName=lie.nextToken();4.4.1求平均值嵌入式系統(tǒng)基本概念

/**獲取學(xué)生成績*/

StringstrScore=lie.nextToken();

word.set(strName);

one.set(Integer.parseInt(strScore));/**輸出姓名和成績*/

context.write(word,one);}}}4.4.1求平均值嵌入式系統(tǒng)基本概念publicstaticclassIntSumReducer

extendsReducer<Text,IntWritable,Text,IntWritable>{privateIntWritableresult=newIntWritable();

publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{4.4.1求平均值嵌入式系統(tǒng)基本概念intsum=0;

intcount=0;for(IntWritable

val:values){/**計(jì)算總分*/sum+=val.get();/**統(tǒng)計(jì)科目數(shù)*/

count++;}4.4.1求平均值嵌入式系統(tǒng)基本概念

/**計(jì)算平均值，并給result賦值*/

result.set((int)(sum/count));

context.write(key,result);}}4.4.1求平均值嵌入式系統(tǒng)基本概念publicstaticvoidmain(String[]args)throwsException{Configurationconf=newConfiguration();String[]otherArgs=newGenericOptionsParser(conf,args).getRemainingArgs();if(otherArgs.length<2){

System.err.println("Usage:wordcount<in>[<in>...]<out>");

System.exit(2);}4.4.1求平均值嵌入式系統(tǒng)基本概念jobjob=Job.getInstance(conf,"wordcount");

job.setJarByClass(WordCount.class);

job.setMapperClass(TokenizerMapper.class);

job.setCombinerClass(IntSumReducer.class);

job.setReducerClass(IntSumReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);4.4.1求平均值嵌入式系統(tǒng)基本概