一 前言

在本系列文章的第一篇：用 Hadoop 進行分布式并行編程，第 1 部分： 基本概念與安裝部署中，介紹了 MapReduce 計算模型，分布式文件系統 HDFS，分布式并行計算等的基本原理, 并且詳細介紹了如何安裝 Hadoop，如何在單機及偽分布式環境　(在一臺單機中用多進程模擬)　運行基于 Hadoop 的并行程序。在本系列文章的第二篇：用 Hadoop 進行分布式并行編程，第 2 部分：程序實例與分析中，介紹了如何針對一個具體的計算任務，基于 Hadoop 編寫 MapReduce 并行程序。 本文將介紹真實的 Hadoop 分布式運行環境, 包括如何在多臺普通的計算機上部署分布式運行環境，如何將 MapReduce 程序遠程部署并運行在這個分布式環境上，并簡略介紹了”云計算平臺” 以及計算能力按需租用服務。

回頁首

二 準備工作

1. 硬件與網絡

使用三臺機器，機器名分別為 homer06, homer07, homer08，均安裝 Redhat Enterprise Linux 5.0 (其它 Linux 發行版亦可), 確保各臺機器之間網絡暢通，機器名與 IP 地址之間解析正確，從任一臺機器都可以　ping 通其它機器的機器名。如有機器名的解析問題，可通過設置 /etc/hosts 文件解決，當然更好的解決方法是在你的網絡中配置 DNS 服務器。此外，需要在三臺機器上創建相同的用戶帳號，如 caoyuz, 或直接使用 root 帳號亦可。

我們將使用 homer06 作為分布式文件系統 HDFS 的 Name Node　及 MapReduce 運行過程中的 Job Tracker 結點，我們將 homer06　稱之為主結點。其它兩臺機器 (homer07, homer08) 作為 HDFS 的 Data Node 以及 MapReduce 運行過程中的 Task Tracker 結點，這些結點可統稱為從結點。如你需要部署更多的機器，也是很容易的，將新加入的機器作為 Data Node　以及 Task Tracker 結點即可，其配置過程與本文介紹的三臺機器的環境類似，此不贅述。

2. SSH 配置

在 Hadoop 分布式環境中，Name Node (主節點) 需要通過 SSH 來啟動和停止 Data Node (從結點)上的各類進程。我們需要保證環境中的各臺機器均可以通過 SSH 登錄訪問，并且 Name Node 用 SSH 登錄 Data Node 時，不需要輸入密碼，這樣 Name Node 才能在后臺自如地控制其它結點。可以將各臺機器上的 SSH 配置為使用無密碼公鑰認證方式來實現。

現在流行的各類 Linux 發行版一般都安裝了 SSH　協議的開源實現 OpenSSH, 并且已經啟動了 SSH 服務, 即這些機器缺省應該就是支持 SSH 登錄的。如果你的機器缺省不支持 SSH, 請下載安裝 OpenSSH。

以下是配置 SSH 的無密碼公鑰認證的過程。首先，在 homer06 機器上執行命令，如代碼清單 1 所示：

代碼清單1

homer06: $ ssh-keygen -t rsa Generating public/private rsa key pair. Enter file in which to save the key (/home/caoyuz/.ssh/id_rsa): Enter passphrase (empty for no passphrase): Enter same passphrase again: Your identification has been saved in /home/caoyuz/.ssh/id_rsa. Your public key has been saved in /home/caoyuz/.ssh/id_rsa.pub. The key fingerprint is: 2e:57:e2:bf:fd:d4:45:5c:a7:51:3d:f1:51:3c:69:68 root@krusty04

這個命令將為 homer06 上的當前用戶 caoyuz 生成其密鑰對，密鑰對的保存路徑使用缺省的 /home/caoyuz/.ssh/id_rsa, 要求輸入 passphrase 的時候，直接回車。這樣生成的證書以及公鑰將存儲在 /home/caoyuz/.ssh 目錄，形成兩個文件 id_rsa，id_rsa.pub。然后將 id_rsa.pub 文件的內容復制到每一臺機器(包括本機 homer06)的　 /home/caoyuz/.ssh/authorized_keys　文件的尾部，如果機器上不存在　/home/caoyuz/.ssh/authorized_keys 文件，可以自行創建一個。請注意 id_rsa.pub 文件的內容是長長的一行，復制時需注意，不要遺漏字符或混入了多余換行符。

接下來可以做一下 SSH 連接測試，從 homer06 分別向 homer06, homer07, homer08 發起 SSH 連接請求，確保不需要輸入密碼就能 SSH 連接成功。 注意第一次 SSH 連接時會出現如下提示信息：

The authenticity of host [homer06] can't be established. The key fingerprint is: 74:32:91:f2:9c:dc:2e:80:48:73:d4:53:ab:e4:d3:1a Are you sure you want to continue connecting (yes/no)?

請輸入 yes, 這樣 OpenSSH 會把連接過來的這臺主機的信息自動加到 /home/caoyuz/.ssh/know_hosts 文件中去，第二次再連接時，就不會有這樣的提示信息了。

回頁首

三 安裝部署 Hadoop

1. 安裝 Hadoop 及 jre1.5

我們首先在主控結點 homer06 上安裝和配置好 Hadoop，安裝過程可以參考本系列文章的第一篇。 假定我們把 Hadoop 安裝在 /home/caoyuz/hadoop-0.16.0目錄中，并且 JRE 1.5 安裝在 /home/caoyuz/jre 目錄下。

2. 修改 conf/hadoop-env.sh 文件

在其中設置 JAVA_HOME 環境變量：export JAVA_HOME=”/home/caoyuz/jre”

3. 修改 conf/hadoop-site.xml 文件

在本系列文章的第一篇中，我們通過修改此文件，配置了 Hadoop 的偽分布式運行模式。現在，我們同樣可以通過配置此文件，配置 Hadoop 的真實的分布式運行環境。請參照代碼清單 2 修改 conf/hadoop-site.xml：

代碼清單2

<configuration> <property> <name>fs.default.name</name> <value>homer06.austin.ibm.com:9000</value> <description>The name of the default file system. Either the literal string "local" or a host:port for DFS.</description> </property> <property> <name>mapred.job.tracker</name> <value>homer06.austin.ibm.com:9001</value> <description>The host and port that the MapReduce job tracker runs at. If "local", then jobs are run in-process as a single map and reduce task.</description> </property> <name>dfs.name.dir</name> <value>/home/caoyuz/hadoopfs/name</value> <description>Determines where on the local filesystem the DFS name node should store the name table. If this is a comma-delimited list of directories then the name table is replicated in all of the directories, for redundancy. </description> </property> <property> <name>dfs.data.dir</name> <value>/home/caoyuz/hadoopfs/data</value> <description>Determines where on the local filesystem an DFS data node should store its blocks. If this is a comma-delimited list of directories, then data will be stored in all named directories, typically on different devices. Directories that do not exist are ignored.</description> </property> <property> <name>dfs.replication</name> <value>2</value> <description>Default block replication. The actual number of replications can be specified when the file is created. The default is used if replication is not specified in create time.</description> </property> </configuration>

參數 fs.default.name 指定 Name Node 的 IP 地址和端口號，此處我們將其設定為 homer06 及 9000 端口，參數 mapred.job.tracker 指定 JobTracker 的 IP 地址和端口號，此處我們將其設定為 homer06 及 9001 端口。 參數 dfs.name.dir 指定 Name Node 相關數據在本地文件系統上的存放位置, 此處我們將其設定為 /home/caoyuz/hadoopfs/name ,參數 dfs.data.dir 指定 Data Node 相關數據在本地文件系統上的存放位置，此處我們將其設定為 /home/caoyuz/hadoopfs/data 。注意, Hadoop 會自動創建這兩個目錄，無需事先創建。

更多的參數配置，可以參考 conf/hadoop-default.xml 文件，并在 conf/hadoop-site.xml 文件中設置。

4. 設定主從節點

修改 conf/masters 文件，將其中的 localhost 改為 homer06 ,修改 conf/slaves 文件, 刪掉其中的 localhost, 將我們的另兩臺機器 homer07, homer08 加入, 注意每個機器一行。

5. 將 Hadoop 部署到其它機器上去

至此, 我們已經在 homer06 上安裝和配置好了 hadoop 和 jre, 現在需要將其部署到其它機器上去，通過 scp 命令即可完成，如代碼清單 3 所示:

代碼清單3

homer06: $ scp -r /home/caoyuz/hadoop-0.16.0 homer07:/home/caoyuz/hadoop-0.16.0 homer06: $ scp -r /home/caoyuz/jre homer07:/home/caoyuz/jre homer06: $ scp -r /home/caoyuz/hadoop-0.16.0 homer08:/home/caoyuz/hadoop-0.16.0 homer06: $ scp -r /home/caoyuz/jre homer08:/home/caoyuz/jre

其中用 scp 拷貝 jre 目錄到其它機器上去不是必須的。你只需保證你的所有機器上均安裝了 JRE1.5 以上版本，并且都是安裝在同一目錄。

6. 在 homer06 上格式化一個新的分布式文件系統

如代碼清單 4 所示:

代碼清單4

homer06: $ cd /home/caoyuz/hadoop-0.16.0 homer06: $ bin/hadoop namenode -format

7. 在 homer06 上啟動 hadoop 進程

如代碼清單5所示:

代碼清單5

homer06: $ cd /home/caoyuz/hadoop-0.16.0 homer06: $ bin/start-all.sh

啟動完成之后，運行 ps -ef 命令應該可以看到 homer06 上啟動了 3 個新的 java 進程 (namenode, secondary namenode, jobtracker), 同時，我們可以到 homer07, homer08 兩臺機器上用 ps –ef 查看，這兩臺機器上應該已經自動啟動了 2 個新的 java 進程 (datanode, tasktracker)

回頁首

四 運行 Hadoop 程序

至此，整個 Hadoop 分布式環境已經部署完畢，并已啟動相關后臺進程?，F在我們可以嘗試運行一下我們在第二篇文章中介紹的 wordcount 程序，如代碼清單 6 所示：

代碼清單 6

homer06: $ mkdir -p /home/test-in # 請先將待測的文件放到本地文件系統的/home/test-in目錄 homer06: $ cd /home/caoyuz/hadoop-0.16.0 homer06: $ bin/hadoop dfs –put /home/test-in input # 將本地文件系統上的 /home/test-in 目錄拷到 HDFS 的根目錄上，目錄名改為 input $ bin/hadoop jar hadoop-0.16.0-examples.jar wordcount input output #查看執行結果: # 將文件從 HDFS 拷到本地文件系統中再查看： $ bin/hadoop dfs -get output output $ cat output/* # 也可以直接查看 $ bin/hadoop dfs -cat output/*

代碼清單 6 所示的執行 wordcount 程序的過程，與我們在第一篇文章中介紹的在偽分布式運行環境運行完全一致，但我們現在擁有了一個真正的分布式執行環境，我們的數據分布存儲于數據節點 homer07 及 homer08 上，可以在這兩臺機器的 /home/caoyuz/hadoopfs/data 目錄 (這是我們在 conf/hadoop-site.xml 中指定的 dfs.data.dir 參數) 下看到一些數據文件，并且整個 wordcount 的計算過程神奇地由 homer06, homer07, homer08 三臺機器并行協同完成，我們還可以很方便的增加更多的機器來參與運算。這就是分布式并行程序的優勢: 可以很容易地通過加入新的機器來獲得更多的存儲空間和計算能力, 部署的機器越多, 就越能有效地完成海量數據的計算。

回頁首

五 使用 IBM MapReduce Tools 部署分布式程序

在第二篇文章中，已經介紹了 IBM MapReduce Tools 的基本功能和用法。現在我們重點介紹如何使用 IBM MapReduce Tools 將 MapReduce 程序遠程部署到 Hadoop 分布式環境中去運行。

假定我們還是使用上一節部署完成的分布式環境，然后是在另一臺機器上使用 Eclipse 開發 MapReduce 程序。

1. 定義 Hadoop server 的位置

首先請確保你的 Eclipse 已經安裝了 IBM MapReduce Tools 這個插件。啟動 Eclipse, 選擇 Window -> Open Perspective ->other, 再從彈出框中選擇 MapReduce, 這樣 Eclipse 會進入專門的 MapReduce 視圖 ( perspective )。

隨后，請檢查你的 MapReduce perspective中是否有一個專門的 MapReduce Servers view, 如果沒有，請選擇 Window -> Show View ->other, 再從彈出框中選擇 MapReduce Tools 類別下面的 MapReduce Servers, 打開這個 view.

然后，請點擊　MapReduce Servers view 右上角的藍色圖標，就會出現如圖一所示的設置 Hadoop Server 的位置的界面。此處所說的 Hadoop server，具體到本文，就是 homer06 這臺機器。在輸入各項參數之后，請點擊 ”Validate location” 按鈕，檢查是否能夠正確的找到并連接上你的 Hadoop server. 如果出錯，請嘗試在命令行下執行命令：ssh the_hostname_of_your_hadoop_server, (或使用圖形界面的 SSH 遠程登錄軟件), 確保 ssh 能夠連接成功。

圖一 定義 Hadoop server 的位置

2. 創立一個 MapReduce Project

在 Eclipse 中新創建一個 MapReduce Project, 將我們在第二篇文章中定義的 WordCount 類加到此 Project 中。這個類需要略作修改才能直接遠程部署到我們已經搭建好的分布式環境中去運行，因為我們原來在　WordCount　程序中是通過讀取命令行參數獲得計算任務的輸入路徑和輸出路徑，而當前版本的 IBM MapReduce Tools 不支持遠程部署時讀取命令行參數。為測試的簡便起見，我在程序中直接將輸入路徑定義為 input, 輸出路徑定義為 output。在測試 WordCount 程序之前，需要事先將需要做詞頻統計的一批文件拷貝到分布式文件系統的 input 目錄下去。

完整的 WordCount 類的代碼如代碼清單 7 所示：

代碼清單7

//import 語句省略 public class WordCount extends Configured implements Tool {public static class MapClass extends MapReduceBaseimplements Mapper<LongWritable, Text, Text, IntWritable> {private final static IntWritable one = new IntWritable(1);private Text word = new Text();private String pattern="[^\\w]";public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {String line = value.toString().toLowerCase();line = line.replaceAll(pattern, " ");StringTokenizer itr = new StringTokenizer(line);while (itr.hasMoreTokens()) {word.set(itr.nextToken());output.collect(word, one);}}}public static class Reduce extends MapReduceBaseimplements Reducer<Text, IntWritable, Text, IntWritable> {public void reduce(Text key, Iterator<IntWritable> values,OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {int sum = 0;while (values.hasNext()) {sum += values.next().get();}output.collect(key, new IntWritable(sum));}}public int run(String[] args) throws Exception {Path tempDir = new Path("wordcount-temp-" + Integer.toString(new Random().nextInt(Integer.MAX_VALUE)));JobConf conf = new JobConf(getConf(), WordCount.class);try {conf.setJobName("wordcount");conf.setOutputKeyClass(Text.class);conf.setOutputValueClass(IntWritable.class);conf.setMapperClass(MapClass.class);conf.setCombinerClass(Reduce.class);conf.setReducerClass(Reduce.class);conf.setInputPath(new Path(args[0]));conf.setOutputPath(tempDir);conf.setOutputFormat(SequenceFileOutputFormat.class);JobClient.runJob(conf);JobConf sortJob = new JobConf(getConf(), WordCount.class);sortJob.setJobName("sort");sortJob.setInputPath(tempDir);sortJob.setInputFormat(SequenceFileInputFormat.class);sortJob.setMapperClass(InverseMapper.class);sortJob.setNumReduceTasks(1); sortJob.setOutputPath(new Path(args[1]));sortJob.setOutputKeyClass(IntWritable.class);sortJob.setOutputValueClass(Text.class);sortJob.setOutputKeyComparatorClass(IntWritableDecreasingComparator.class);JobClient.runJob(sortJob);} finally {FileSystem.get(conf).delete(tempDir);}return 0;}private static class IntWritableDecreasingComparator extends IntWritable.Comparator {public int compare(WritableComparable a, WritableComparable b) {return -super.compare(a, b);}public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {return -super.compare(b1, s1, l1, b2, s2, l2);}}public static void main(String[] args) throws Exception {String[] paths = {"input" , "output"};int res = ToolRunner.run(new Configuration(), new WordCount(), paths);System.exit(res);} }

3. 遠程部署與運行

在左側的 Project Explorer 中選中 WordCount 類，在右鍵彈出菜單中選擇 Run As->Run on hadoop, 如圖二 所示：

圖二

然后在 “select hadoop server” 彈出框中選擇我們已經定義好的 Hadoop server, 點擊 Finish 之后，MapReduce Tool 會自動將 WordCount project打包成一個 jar 并拷到遠程 Hadoop server 上運行起來, 整個運行過程的輸出在 Eclipse 的 console 中即可看到，非常方便。

4. 查看運行結果

當定義好 Hadoop server 的位置之后，在左側的 Project Explorer 會出現一個新的 project( 項目名前面有一個藍色的小象圖標), , 通過這個 project 可以瀏覽 Hadoop 分布式文件系統中的文件。雙擊 output 目錄下的 part-0000 文件，我們就可以直接在 Eclipse 中查看 WordCount 程序的輸出結果，如圖三所示：

圖三

回頁首

六 云計算與 Hadoop

我們知道，在分布式集群環境中才能發揮 Hadoop 的并行優勢，擁有的機器數量越多，越能快速有效的處理海量數據?，F實問題是，雖然很多公司都有處理海量數據的需求，卻又不可能專門投資去搭建大規模的集群環境，Hadoop 于他們，不免淪為”屠龍之技”，無處發揮其優勢，如之奈何？在過去，這個問題還真是難以解決，今天的情況就不一樣了。讀者如果關注 IT 業界動態，當知現在 IT 業界正在極力鼓吹”云計算”, 并有一些公司開始投資搭建所謂的”云計算平臺”，這里的”云”, 就是一堆機器組成的分布式環境外加一些基礎構架軟件和管理軟件，其中便會有類似于 Hadoop 這樣的分布式計算軟件，HDFS 這樣的分布式文件系統，有需求的公司和個人可以到這樣的”云計算平臺”上去租用存儲空間，租用計算結點(計算能力)做分布式運算。

比如 Amazon 公司基于 Hadoop 推出了 Amazon S3 ( Amazon Simple Storage Service )，提供可靠，快速，可擴展的網絡存儲服務，以及一個商用的云計算平臺 Amazon EC2 ( Amazon Elastic Compute Cloud )。用戶可以將其數據存儲在 Amazon S3 分布式存儲平臺上， 然后到 Amazon EC2 上去租用計算能力，完成對數據的計算。Amazon EC2 提供所謂的按需租用服務，目前的收費標準是每臺虛擬計算機 (Amazon EC2 稱之為一個 instance) 每小時0.10美元。與傳統的主機租用服務完全不同，用戶可以根據自己某次運算處理的規模，租用相應數量的虛擬計算機，運算完畢后就可以釋放你租用的虛擬計算機，Amazon 則會根據你租用的虛擬計算機的數量以及本次計算的實際運行時間向你收費，等于說你花錢租用計算能力，但不會浪費一個子兒。IBM 公司的云計算平臺"藍云"也面向企業用戶提供了類似的功能。

如果我們打算基于 Hadoop 編寫分布式并行程序來處理大量的數據，完全可以到 IBM, Amazon 等提供的云計算平臺上去進行計算，對于 IBM 藍云，Amazon S3, Amazon EC2 的詳細介紹超出了本文范圍，有興趣的讀者可以去其官方網站了解更多的信息。

回頁首

七 結束語

這是系列文章的最后一篇。第一篇文章介紹了 MapReduce 計算模型，分布式文件系統 HDFS，分布式并行計算等的基本原理, 如何安裝和部署單機 Hadoop 環境, 在第二篇文章中，我們實際編寫了一個 Hadoop 并行計算程序，并了解了一些重要的編程細節，了解了如何使用 IBM MapReduce Tools 在 Eclipse 環境中編譯，運行和調試 Hadoop 并行計算程序。本篇文章則詳細介紹了如何部署分布式 Hadoop 環境，如何利用 IBM MapReduce Tools 將程序部署到分布式環境中運行，并簡略介紹了現在流行的”云計算平臺” 以及計算能力按需租用服務。

希望這三篇文章能起到一個拋磚引玉的作用，讓你感受到 MapReduce 分布式并行編程的樂趣并從此入門且樂在其中，為即將到來的所謂”云計算”時代提前熱熱身。

聲明：本文僅代表作者個人之觀點，不代表 IBM 公司之觀點。

參考資料

學習

• 訪問?Hadoop 官方網站,了解 Hadoop 及其子項目 HBase 的信息。
• Hadoop wiki上, 有許多 Hadoop 的用戶文檔,開發文檔,示例程序等。
• 閱讀 Google Mapduce 論文：?MapReduce: Simplified Data Processing on Large Clusters, 深入了解 Mapreduce 計算模型。
• 深入了解 Hadoop 分布式文件系統 HDFS：?The Hadoop Distributed File System:Architecture and Design
• 深入了解 Google 文件系統 GFS:?The Google File System, Hadoop HDFS 實現了與 GFS 類似的功能。
• IBM MapReduce Tools:?http://www.alphaworks.ibm.com/tech/mapreducetools,

from:?http://www.ibm.com/developerworks/cn/opensource/os-cn-hadoop3/index.html

總結

以上是生活随笔為你收集整理的用 Hadoop 进行分布式并行编程, 第 3 部分 部署到分布式环境的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。