Hadoop 附帶了一個(gè)名為 HDFS(Hadoop分布式文件系統(tǒng))的分布式文件系統(tǒng)，基于 Hadoop 的應(yīng)用程序使用 HDFS 。HDFS 是專為存儲(chǔ)超大數(shù)據(jù)文件，運(yùn)行在集群的商品硬件上。它是容錯(cuò)的，可伸縮的，并且非常易于擴(kuò)展。 你知道嗎? ?當(dāng)數(shù)據(jù)超過(guò)一個(gè)單個(gè)物理機(jī)器上存儲(chǔ)的容量，除以跨獨(dú)立機(jī)器數(shù)。管理跨越機(jī)器的網(wǎng)絡(luò)存儲(chǔ)特定操作被稱為分布式文件系統(tǒng)。

HDFS集群主要由 NameNode 管理文件系統(tǒng) Metadata 和 DataNodes 存儲(chǔ)的實(shí)際數(shù)據(jù)。

• NameNode:?NameNode可以被認(rèn)為是系統(tǒng)的主站。它維護(hù)所有系統(tǒng)中存在的文件和目錄的文件系統(tǒng)樹(shù)和元數(shù)據(jù) 。 兩個(gè)文件：“命名空間映像“和”編輯日志“是用來(lái)存儲(chǔ)元數(shù)據(jù)信息。Namenode 有所有包含數(shù)據(jù)塊為一個(gè)給定的文件中的數(shù)據(jù)節(jié)點(diǎn)的知識(shí)，但是不存儲(chǔ)塊的位置持續(xù)。從數(shù)據(jù)節(jié)點(diǎn)在系統(tǒng)每次啟動(dòng)時(shí)信息重構(gòu)一次。
• DataNode :?DataNodes作為從機(jī)，每臺(tái)機(jī)器位于一個(gè)集群中，并提供實(shí)際的存儲(chǔ). 它負(fù)責(zé)為客戶讀寫請(qǐng)求服務(wù)。

HDFS中的讀/寫操作運(yùn)行在塊級(jí)。HDFS數(shù)據(jù)文件被分成塊大小的塊，這是作為獨(dú)立的單元存儲(chǔ)。默認(rèn)塊大小為64 MB。

HDFS操作上是數(shù)據(jù)復(fù)制的概念，其中在數(shù)據(jù)塊的多個(gè)副本被創(chuàng)建，分布在整個(gè)節(jié)點(diǎn)的群集以使在節(jié)點(diǎn)故障的情況下數(shù)據(jù)的高可用性。

注： ?在HDFS的文件，比單個(gè)塊小，不占用塊的全部存儲(chǔ)。

在HDFS讀操作

數(shù)據(jù)讀取請(qǐng)求將由 HDFS，NameNode和DataNode來(lái)服務(wù)。讓我們把讀取器叫 “客戶”。下圖描繪了文件的讀取操作在 Hadoop 中。

客戶端啟動(dòng)通過(guò)調(diào)用文件系統(tǒng)對(duì)象的 open() 方法讀取請(qǐng)求; 它是?DistributedFileSystem?類型的對(duì)象。

此對(duì)象使用 RPC 連接到 namenode 并獲取的元數(shù)據(jù)信息，如該文件的塊的位置。 請(qǐng)注意，這些地址是文件的前幾個(gè)塊。

響應(yīng)該元數(shù)據(jù)請(qǐng)求，具有該塊副本的 DataNodes 地址被返回。

一旦接收到 DataNodes 的地址，FSDataInputStream 類型的一個(gè)對(duì)象被返回到客戶端。?FSDataInputStream 包含 DFSInputStream 這需要處理交互 DataNode 和 NameNode。在上圖所示的步驟4，客戶端調(diào)用 read() 方法，這將導(dǎo)致 DFSInputStream 建立與第一個(gè) DataNode 文件的第一個(gè)塊連接。

以數(shù)據(jù)流的形式讀取數(shù)據(jù)，其中客戶端多次調(diào)用 “read() ” 方法。 read() 操作這個(gè)過(guò)程一直持續(xù)，直到它到達(dá)塊結(jié)束位置。

一旦到模塊的結(jié)尾，DFSInputStream 關(guān)閉連接，移動(dòng)定位到下一個(gè) DataNode 的下一個(gè)塊

一旦客戶端已讀取完成后，它會(huì)調(diào)用 close()方法。

HDFS寫操作

在本節(jié)中，我們將了解如何通過(guò)的文件將數(shù)據(jù)寫入到 HDFS。

客戶端通過(guò)調(diào)用 DistributedFileSystem對(duì)象的 create() 方法創(chuàng)建一個(gè)新的文件，并開(kāi)始寫操作 - 在上面的圖中的步驟1

DistributedFileSystem對(duì)象使用 RPC 調(diào)用連接到 NameNode，并啟動(dòng)新的文件創(chuàng)建。但是，此文件創(chuàng)建操作不與文件任何塊相關(guān)聯(lián)。NameNode 的責(zé)任是驗(yàn)證文件(其正被創(chuàng)建的)不存在，并且客戶端具有正確權(quán)限來(lái)創(chuàng)建新文件。如果文件已經(jīng)存在，或者客戶端不具有足夠的權(quán)限來(lái)創(chuàng)建一個(gè)新的文件，則拋出 IOException 到客戶端。否則操作成功，并且該文件新的記錄是由 NameNode 創(chuàng)建。

一旦 NameNode 創(chuàng)建一條新的記錄，返回FSDataOutputStream 類型的一個(gè)對(duì)象到客戶端。客戶端使用它來(lái)寫入數(shù)據(jù)到 HDFS。數(shù)據(jù)寫入方法被調(diào)用(圖中的步驟3)。

FSDataOutputStream包含DFSOutputStream對(duì)象，它使用 DataNodes 和 NameNode 通信后查找。當(dāng)客戶機(jī)繼續(xù)寫入數(shù)據(jù)，DFSOutputStream?繼續(xù)創(chuàng)建這個(gè)數(shù)據(jù)包。這些數(shù)據(jù)包連接排隊(duì)到一個(gè)隊(duì)列被稱為?DataQueue

還有一個(gè)名為 DataStreamer 組件，用于消耗DataQueue。DataStreamer 也要求 NameNode 分配新的塊，揀選 DataNodes 用于復(fù)制。

現(xiàn)在，復(fù)制過(guò)程始于使用 DataNodes 創(chuàng)建一個(gè)管道。 在我們的例子中，選擇了復(fù)制水平3，因此有 3 個(gè) DataNodes 管道。

所述 DataStreamer 注入包分成到第一個(gè) DataNode 的管道中。

在每個(gè) DataNode 的管道中存儲(chǔ)數(shù)據(jù)包接收并同樣轉(zhuǎn)發(fā)在第二個(gè) DataNode 的管道中。

另一個(gè)隊(duì)列，“Ack Queue”是由 DFSOutputStream 保持存儲(chǔ)，它們是 DataNodes 等待確認(rèn)的數(shù)據(jù)包。

一旦確認(rèn)在隊(duì)列中的分組從所有 DataNodes 已接收在管道，它從 'Ack Queue' 刪除。在任何 DataNode 發(fā)生故障時(shí)，從隊(duì)列中的包重新用于操作。

在客戶端的數(shù)據(jù)寫入完成后，它會(huì)調(diào)用close()方法(第9步圖中)，調(diào)用close()結(jié)果進(jìn)入到清理緩存剩余數(shù)據(jù)包到管道之后等待確認(rèn)。

一旦收到最終確認(rèn)，NameNode 連接告訴它該文件的寫操作完成。

使用JAVA API訪問(wèn)HDFS

在本節(jié)中，我們來(lái)了解 Java 接口并用它們來(lái)訪問(wèn)Hadoop的文件系統(tǒng)。

為了使用編程方式與 Hadoop 文件系統(tǒng)進(jìn)行交互，Hadoop 提供多種 Java 類。org.apache.hadoop.fs包中包含操縱 Hadoop 文件系統(tǒng)中的文件類工具。這些操作包括，打開(kāi)，讀取，寫入，和關(guān)閉。實(shí)際上，對(duì)于 Hadoop 文件 API 是通用的，可以擴(kuò)展到 HDFS 的其他文件系統(tǒng)交互。

編程從 HDFS 讀取文件

java.net.URL 對(duì)象是用于讀取文件的內(nèi)容。首先，我們需要讓 Java 識(shí)別 Hadoop 的 HDFS URL架構(gòu)。這是通過(guò)調(diào)用 URL 對(duì)象的?setURLStreamHandlerFactory方法和?FsUrlStreamHandlerFactory的一個(gè)實(shí)例琮傳遞給它。此方法只需要執(zhí)行一次在每個(gè)JVM，因此，它被封閉在一個(gè)靜態(tài)塊中。

示例代碼

?

publicclassURLCat { static{ URL.setURLStreamHandlerFactory(newFsUrlStreamHandlerFactory()); } publicstaticvoidmain(String[] args)?throwsException { InputStream in =?null; try{ in =?newURL(args[0]).openStream(); IOUtils.copyBytes(in, System.out,?4096,?false); }?finally{ IOUtils.closeStream(in); } } }

這段代碼用于打開(kāi)和讀取文件的內(nèi)容。HDFS文件的路徑作為命令行參數(shù)傳遞給該程序。

使用命令行界面訪問(wèn)HDFS

這是與 HDFS 交互的最簡(jiǎn)單的方法之一。 命令行接口支持對(duì)文件系統(tǒng)操作，例如：如讀取文件，創(chuàng)建目錄，移動(dòng)文件，刪除數(shù)據(jù)，并列出目錄。

可以執(zhí)行?'$HADOOP_HOME/bin/hdfs dfs -help'?來(lái)獲得每一個(gè)命令的詳細(xì)幫助。這里,?'dfs'?HDFS是一個(gè)shell命令，它支持多個(gè)子命令。首先要啟動(dòng) Haddop 服務(wù)(使用 hduser_用戶)，執(zhí)行命令如下：

hduser_@ubuntu:~$ su hduser_ hduser_@ubuntu:~$ $HADOOP_HOME/sbin/start-dfs.sh hduser_@ubuntu:~$ $HADOOP_HOME/sbin/start-yarn.sh

一些廣泛使用的命令的列表如下

1. 從本地文件系統(tǒng)復(fù)制文件到 HDFS

hduser_@ubuntu:~$ $HADOOP_HOME/bin/hdfs dfs -copyFromLocal temp.txt /

此命令將文件從本地文件系統(tǒng)拷貝 temp.txt 文件到 HDFS。

2. 我們可以通過(guò)以下命令列出一個(gè)目錄下存在的文件?-ls

hduser_@ubuntu:~$ $HADOOP_HOME/bin/hdfs dfs -ls /

我們可以看到一個(gè)文件 'temp.txt“(之前復(fù)制)被列在”/“目錄。

3. 以下命令將文件從 HDFS 拷貝到本地文件系統(tǒng)

hduser_@ubuntu:~$ $HADOOP_HOME/bin/hdfs dfs -copyToLocal /temp.txt

我們可以看到 temp.txt 已經(jīng)復(fù)制到本地文件系統(tǒng)。

4. 以下命令用來(lái)創(chuàng)建新的目錄

hduser_@ubuntu:~$ $HADOOP_HOME/bin/hdfs dfs -mkdir /mydirectory

接下來(lái)檢查是否已經(jīng)建立了目錄。現(xiàn)在，應(yīng)該知道怎么做了吧？

MapReduce 是適合海量數(shù)據(jù)處理的編程模型。Hadoop是能夠運(yùn)行在使用各種語(yǔ)言編寫的MapReduce程序：?Java, Ruby, Python, and C++. MapReduce程序是平行性的，因此可使用多臺(tái)機(jī)器集群執(zhí)行大規(guī)模的數(shù)據(jù)分析非常有用的。

MapReduce程序的工作分兩個(gè)階段進(jìn)行：

Map階段

Reduce 階段

輸入到每一個(gè)階段均是鍵 - 值對(duì)。此外，每一個(gè)程序員需要指定兩個(gè)函數(shù)：map函數(shù)和reduce函數(shù)

整個(gè)過(guò)程要經(jīng)歷三個(gè)階段執(zhí)行，即

MapReduce如何工作

讓我們用一個(gè)例子來(lái)理解這一點(diǎn) –

假設(shè)有以下的輸入數(shù)據(jù)到 MapReduce 程序，統(tǒng)計(jì)以下數(shù)據(jù)中的單詞數(shù)量：

Welcome to Hadoop Class

Hadoop is good

Hadoop is bad

MapReduce 任務(wù)的最終輸出是：

bad?	1
Class?	1
good?	1
Hadoop?	3
is?	2
to?	1
Welcome?	1

這些數(shù)據(jù)經(jīng)過(guò)以下幾個(gè)階段

輸入拆分：

輸入到MapReduce工作被劃分成固定大小的塊叫做 input splits ，輸入折分是由單個(gè)映射消費(fèi)輸入塊。

映射 - Mapping

這是在 map-reduce 程序執(zhí)行的第一個(gè)階段。在這個(gè)階段中的每個(gè)分割的數(shù)據(jù)被傳遞給映射函數(shù)來(lái)產(chǎn)生輸出值。在我們的例子中，映射階段的任務(wù)是計(jì)算輸入分割出現(xiàn)每個(gè)單詞的數(shù)量(更多詳細(xì)信息有關(guān)輸入分割在下面給出)并編制以某一形式列表<單詞，出現(xiàn)頻率>

重排

這個(gè)階段消耗映射階段的輸出。它的任務(wù)是合并映射階段輸出的相關(guān)記錄。在我們的例子，同樣的詞匯以及它們各自出現(xiàn)頻率。

Reducing

在這一階段，從重排階段輸出值匯總。這個(gè)階段結(jié)合來(lái)自重排階段值，并返回一個(gè)輸出值。總之，這一階段匯總了完整的數(shù)據(jù)集。

在我們的例子中，這個(gè)階段匯總來(lái)自重排階段的值，計(jì)算每個(gè)單詞出現(xiàn)次數(shù)的總和。

詳細(xì)的整個(gè)過(guò)程

• 映射的任務(wù)是為每個(gè)分割創(chuàng)建在分割每條記錄執(zhí)行映射的函數(shù)。
• 有多個(gè)分割是好處的， 因?yàn)樘幚硪粋€(gè)分割使用的時(shí)間相比整個(gè)輸入的處理的時(shí)間要少， 當(dāng)分割比較小時(shí)，處理負(fù)載平衡是比較好的，因?yàn)槲覀冋诓⑿械靥幚矸指睢?
• 然而，也不希望分割的規(guī)模太小。當(dāng)分割太小，管理分割和映射創(chuàng)建任務(wù)的超負(fù)荷開(kāi)始逐步控制總的作業(yè)執(zhí)行時(shí)間。
• 對(duì)于大多數(shù)作業(yè)，最好是分割成大小等于一個(gè)HDFS塊的大小(這是64 MB，默認(rèn)情況下)。
• map任務(wù)執(zhí)行結(jié)果到輸出寫入到本地磁盤的各個(gè)節(jié)點(diǎn)上，而不是HDFS。
• 之所以選擇本地磁盤而不是HDFS是因?yàn)?#xff0c;避免復(fù)制其中發(fā)生 HDFS 存儲(chǔ)操作。
• 映射輸出是由減少任務(wù)處理以產(chǎn)生最終的輸出中間輸出。
• 一旦任務(wù)完成，映射輸出可以扔掉了。所以，復(fù)制并將其存儲(chǔ)在HDFS變得大材小用。
• 在節(jié)點(diǎn)故障的映射輸出之前，由 reduce 任務(wù)消耗，Hadoop 重新運(yùn)行另一個(gè)節(jié)點(diǎn)在映射上的任務(wù)，并重新創(chuàng)建的映射輸出。
• 減少任務(wù)不會(huì)在數(shù)據(jù)局部性的概念上工作。每個(gè)map任務(wù)的輸出被供給到 reduce 任務(wù)。映射輸出被傳輸至計(jì)算機(jī)，其中 reduce 任務(wù)正在運(yùn)行。
• 在此機(jī)器輸出合并，然后傳遞到用戶定義的 reduce 函數(shù)。
• 不像到映射輸出，reduce輸出存儲(chǔ)在HDFS(第一個(gè)副本被存儲(chǔ)在本地節(jié)點(diǎn)上，其他副本被存儲(chǔ)于偏離機(jī)架的節(jié)點(diǎn))。因此，寫入 reduce 輸出

MapReduce如何組織工作？

Hadoop 劃分工作為任務(wù)。有兩種類型的任務(wù)：

Map 任務(wù)?(分割及映射)

Reduce 任務(wù)?(重排，還原)

如上所述

完整的執(zhí)行流程(執(zhí)行 Map 和 Reduce 任務(wù))是由兩種類型的實(shí)體的控制，稱為

Jobtracker?: 就像一個(gè)主(負(fù)責(zé)提交的作業(yè)完全執(zhí)行)

多任務(wù)跟蹤器?: 充當(dāng)角色就像從機(jī)，它們每個(gè)執(zhí)行工作

對(duì)于每一項(xiàng)工作提交執(zhí)行在系統(tǒng)中，有一個(gè) JobTracker 駐留在 Namenode 和 Datanode 駐留多個(gè) TaskTracker。

• 作業(yè)被分成多個(gè)任務(wù)，然后運(yùn)行到集群中的多個(gè)數(shù)據(jù)節(jié)點(diǎn)。
• JobTracker的責(zé)任是協(xié)調(diào)活動(dòng)調(diào)度任務(wù)來(lái)在不同的數(shù)據(jù)節(jié)點(diǎn)上運(yùn)行。
• 單個(gè)任務(wù)的執(zhí)行，然后由 TaskTracker 處理，它位于執(zhí)行工作的一部分，在每個(gè)數(shù)據(jù)節(jié)點(diǎn)上。
• TaskTracker 的責(zé)任是發(fā)送進(jìn)度報(bào)告到JobTracker。
• 此外，TaskTracker 周期性地發(fā)送“心跳”信號(hào)信息給 JobTracker 以便通知系統(tǒng)它的當(dāng)前狀態(tài)。
• 這樣 JobTracker 就可以跟蹤每項(xiàng)工作的總體進(jìn)度。在任務(wù)失敗的情況下，JobTracker 可以在不同的 TaskTracker 重新調(diào)度它。

問(wèn)題陳述：

找出銷往各個(gè)國(guó)家商品數(shù)量。

輸入:?我們的畋輸入數(shù)據(jù)集合是一個(gè) CSV 文件,?Sales2014.csv

前提條件：

• 本教程是在Linux上開(kāi)發(fā) - Ubuntu操作系統(tǒng)
• 已經(jīng)安裝了Hadoop(本教程使用版本2.7.1)
• 系統(tǒng)上已安裝了Java(本教程使用 JDK1.8.0)。

在實(shí)際操作過(guò)程中，使用的用戶是'hduser_“(此用戶使用 Hadoop)。

yiibai@ubuntu:~$ su hduser_

步驟:

1.創(chuàng)建一個(gè)新的目錄名稱是：MapReduceTutorial

hduser_@ubuntu:~$ sudo mkdir MapReduceTuorial

授予權(quán)限

hduser_@ubuntu:~$ sudo chmod -R 777 MapReduceTutorial

下載相關(guān)文件：下載 Java 程序文件，拷貝以下文件：SalesMapper.java, SalesCountryReducer.java?和?SalesCountryDriver.java?到 MapReduceTutorial 目錄中，

檢查所有這些文件的文件權(quán)限是否正確：

如果“讀取”權(quán)限缺少可重新再授予權(quán)限，執(zhí)行以下命令：

yiibai@ubuntu:/home/hduser_/MapReduceTutorial$ sudo chmod +r *

2.導(dǎo)出類路徑

hduser_@ubuntu:~/MapReduceTutorial$ export CLASSPATH="$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.7.1.jar:$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-client-common-2.7.1.jar:$HADOOP_HOME/share/hadoop/common/hadoop-common-2.7.1.jar:~/MapReduceTutorial/SalesCountry/*:$HADOOP_HOME/lib/*" hduser_@ubuntu:~/MapReduceTutorial$

3.?編譯Java文件(這些文件存在于目錄：Final-MapReduceHandsOn). 它的類文件將被放在包目錄：

hduser_@ubuntu:~/MapReduceTutorial$ javac -d . SalesMapper.java SalesCountryReducer.java SalesCountryDriver.java

安全地忽略此警告：

此編譯將創(chuàng)建一個(gè)名稱與Java源文件(在我們的例子即，SalesCountry)指定包名稱的目錄，并把所有編譯的類文件在里面，因此這個(gè)目錄要在編譯文件前創(chuàng)建。

接下來(lái)：

創(chuàng)建一個(gè)新的文件：Manifest.txt

hduser_@ubuntu:~/MapReduceTutorial$ vi Manifest.txt

添加以下內(nèi)容到文件中：

Main-Class: SalesCountry.SalesCountryDriver

SalesCountry.SalesCountryDriver?是主類的名稱。請(qǐng)注意，必須鍵入回車鍵，在該行的末尾。

下一步：創(chuàng)建一個(gè) jar 文件

hduser_@ubuntu:~/MapReduceTutorial$ $JAVA_HOME/bin/jar cfm ProductSalePerCountry.jar Manifest.txt SalesCountry/*.class

檢查所創(chuàng)建的 jar 文件，結(jié)果如下：

6.?啟動(dòng) Hadoop

hduser_@ubuntu:~$ $HADOOP_HOME/sbin/start-dfs.sh hduser_@ubuntu:~$ $HADOOP_HOME/sbin/start-yarn.sh

7.?拷貝文件?Sales2014.csv?到?~/inputMapReduce

hduser_@ubuntu:~$ mkdir inputMapReduce hduser_@ubuntu:~$ cp MapReduceTutorial/Sales2014.csv ./inputMapReduce/Sales2014.csv

現(xiàn)在使用以下命令來(lái)拷貝?~/inputMapReduce?到?HDFS.

hduser_@ubuntu:~$ $HADOOP_HOME/bin/hdfs dfs -copyFromLocal ~/inputMapReduce /

我們可以放心地忽略此警告。驗(yàn)證文件是否真正復(fù)制沒(méi)有？

hduser_@ubuntu:~$ $HADOOP_HOME/bin/hdfs dfs -ls /inputMapReduce

8.?運(yùn)行MapReduce 作業(yè)

hduser_@ubuntu:~$ $HADOOP_HOME/bin/hadoop jar ProductSalePerCountry.jar /inputMapReduce /mapreduce_output_sales

這將在 HDFS 上創(chuàng)建一個(gè)輸出目錄，名為mapreduce_output_sales。此目錄的文件內(nèi)容將包含每個(gè)國(guó)家的產(chǎn)品銷售。

9.?結(jié)果可以通過(guò)命令界面中可以看到

hduser_@ubuntu:~$ $HADOOP_HOME/bin/hdfs dfs -cat /mapreduce_output_sales/part-00000

結(jié)果也可以通過(guò) Web 界面看到，打開(kāi) Web 瀏覽器，輸入網(wǎng)址：http://localhost:50070/dfshealth.jsp ，結(jié)果如下：

現(xiàn)在選擇?'Browse the filesystem'?并導(dǎo)航到?/mapreduce_output_sales?如下：

打開(kāi)?part-r-00000 ，如下圖所示：

下載后，查看結(jié)果內(nèi)容。

from: http://www.yiibai.com/hadoop/

總結(jié)

以上是生活随笔為你收集整理的Hadoop教程（三）：HDFS、MapReduce、程序入门实践的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。