使用Cygwin模擬Linux環(huán)境安裝配置運(yùn)行基于單機(jī)的Hadoop其實(shí)，使用Cygwin模擬Linux環(huán)境來(lái)運(yùn)行Hadoop是非常輕松的，只需要簡(jiǎn)單地配置一下就可以運(yùn)行基于單機(jī)的Hadoop。這里，比較關(guān)鍵的就是Cygwin的安裝，在選擇安裝的時(shí)候一定要安裝openssh，否則不會(huì)成功的，下面簡(jiǎn)單說(shuō)一下Cygwin的安裝及其配置：Cygwin的下載安裝首先點(diǎn)擊/setup.exe下載setup.exe，例如保存到桌面，點(diǎn)擊就可以進(jìn)行下載安裝了。在選擇安裝類型的時(shí)候，最好是選擇第一個(gè)，直接從網(wǎng)絡(luò)上下載并緊接著安裝，如圖所示：然后選擇安裝路徑、安裝文件存放路徑、連接方式(這里選擇UseIE5Settings)、下載站點(diǎn)鏡像，自動(dòng)創(chuàng)建下載文件列表，接下來(lái)一步比較重要了：選擇安裝類型，可以單擊最頂層的All后面的循環(huán)樣式圖標(biāo)切換安裝類型，是的最頂層All行的最后一個(gè)單詞為Install，如圖所示：其實(shí)，如果你選擇了Install安裝類型，就已經(jīng)選擇了openssh包。為讓你看到openssh包，你可以在Net[圖標(biāo)]Install下面看到與網(wǎng)絡(luò)有關(guān)的包，如圖所示：向下滑動(dòng)滾動(dòng)條，可以看到openssh，如圖所示：在Cirrent下如果顯示版本號(hào)，說(shuō)明該包已經(jīng)被此次安裝選擇上了，否則的話會(huì)顯示一個(gè)Skip，意思是跳過(guò)該包，并不會(huì)安裝該包的。最后就等著下載安裝了，這個(gè)過(guò)程可能會(huì)花費(fèi)一點(diǎn)時(shí)間的。Cygwin的配置安裝完成之后，例如我的Cygwin安裝在G:\Cygwin\目錄下面，進(jìn)行配置如下：設(shè)置環(huán)境變量：在系統(tǒng)變量中新建變量【變量名：CYGWIN，變量值：ntsectty】；編輯添加變量【變量名：Path，變量值：G:\Cygwin\bin;其它的保留】。OK，基本配置好了，可以配置Hadoop了。Hadoop目前的有幾個(gè)版本：hadoop-0.16.4、hadoop-0.18.0，到Apache下載一個(gè)并解壓縮即可。將解壓縮的Hadoop放到G盤下，例如我的是：G:\hadoop-0.16.4。配置Hadoop只需要修改G:\hadoop-0.16.4\conf目錄下的hadoop-env.sh文件即可，打開(kāi)它你可以看到：#Thejavaimplementationtouse.Required.

#exportJAVA_HOME=/usr/lib/j2sdk1.5-sun將第二行的注釋符號(hào)去掉，同時(shí)指定在你的機(jī)器上JAVA_HOME的值，如下為我修改的內(nèi)容：#Thejavaimplementationtouse.Required.

exportJAVA_HOME="D:\ProgramFiles\Java\jdk1.6.0_07"這里要注意，如果你的JDK安裝目錄中存在空格，需要使用雙引號(hào)引起來(lái)，否則就會(huì)報(bào)錯(cuò)。啟動(dòng)Cygwin，當(dāng)前它是在home/yourname目錄下的，如圖所示：切換到根目錄下，從而進(jìn)入G:\hadoop-0.16.4目錄，并創(chuàng)建一個(gè)數(shù)據(jù)輸入目錄input-dir，如圖所示：下面，打開(kāi)G:\hadoop-0.16.4\input-dir，在該目錄下新建幾個(gè)記事本文件，例如我創(chuàng)建了3個(gè)：input-a.txt、input-b.txt、input-c.txt。三個(gè)文件的內(nèi)容如下所示：input-a.txt：asafterappendactorasasaddasapacheasafteraddasinput-b.txt：benchbebenchbelievebackgroundbenchbeblockinput-c.txt：cafecatcommunicationsconnectioncatcatcatcustcafe接著就可以執(zhí)行Hadoop自帶的一個(gè)統(tǒng)計(jì)英文單詞出現(xiàn)頻率的例子了，直接輸入命令bin/hadoopjarhadoop-0.16.4-examples.jarwordcountinput-diroutput-dir，其中hadoop-0.16.4-examples.jar是G:\hadoop-0.16.4包中的例子，input-dir是數(shù)據(jù)輸入目錄，在里面已經(jīng)存在我們創(chuàng)建的三個(gè)文件了，output-dir是經(jīng)過(guò)Hadoop處理后的輸出結(jié)果的目錄，這里，你需要對(duì)Google的MapReduce算法有一個(gè)簡(jiǎn)單的了解，主要就是在處理數(shù)據(jù)的時(shí)候是怎樣的一個(gè)流程，引用IBM上的一片技術(shù)文章片段了解一下：引用用MapReduce來(lái)處理大數(shù)據(jù)集的過(guò)程,這個(gè)MapReduce的計(jì)算過(guò)程簡(jiǎn)而言之，就是將大數(shù)據(jù)集分解為成百上千的小數(shù)據(jù)集，每個(gè)(或若干個(gè))數(shù)據(jù)集分別由集群中的一個(gè)結(jié)點(diǎn)(一般就是一臺(tái)普通的計(jì)算機(jī))進(jìn)行處理并生成中間結(jié)果，然后這些中間結(jié)果又由大量的結(jié)點(diǎn)進(jìn)行合并,形成最終結(jié)果。計(jì)算模型的核心是Map和Reduce兩個(gè)函數(shù)，這兩個(gè)函數(shù)由用戶負(fù)責(zé)實(shí)現(xiàn)，功能是按一定的映射規(guī)則將輸入的<key,value>對(duì)轉(zhuǎn)換成另一個(gè)或一批<key,value>對(duì)輸出。

表一Map和Reduce函數(shù)函數(shù)輸入輸出說(shuō)明Map<k1,v1>List(<k2,v2>)1.將小數(shù)據(jù)集進(jìn)一步解析成一批<key,value>對(duì)，輸入Map函數(shù)中進(jìn)行處理。

2.每一個(gè)輸入的<k1,v1>會(huì)輸出一批<k2,v2>。<k2,v2>是計(jì)算的中間結(jié)果。Reduce<k2,List(v2)><k3,v3>輸入的中間結(jié)果<k2,List(v2)>中的List(v2)表示是一批屬于同一個(gè)k2的value基于MapReduce計(jì)算模型編寫分布式并行程序非常簡(jiǎn)單，程序員的主要編碼工作就是實(shí)現(xiàn)Map和Reduce函數(shù)，其它的并行編程中的種種復(fù)雜問(wèn)題，如分布式存儲(chǔ)，工作調(diào)度，負(fù)載平衡，容錯(cuò)處理，網(wǎng)絡(luò)通信等，均由MapReduce框架(比如Hadoop)負(fù)責(zé)處理，程序員完全不用操心。本地計(jì)算數(shù)據(jù)存儲(chǔ)在哪一臺(tái)計(jì)算機(jī)上，就由這臺(tái)計(jì)算機(jī)進(jìn)行這部分?jǐn)?shù)據(jù)的計(jì)算，這樣可以減少數(shù)據(jù)在網(wǎng)絡(luò)上的傳輸，降低對(duì)網(wǎng)絡(luò)帶寬的需求。在Hadoop這樣的基于集群的分布式并行系統(tǒng)中，計(jì)算結(jié)點(diǎn)可以很方便地?cái)U(kuò)充，而因它所能夠提供的計(jì)算能力近乎是無(wú)限的，但是由是數(shù)據(jù)需要在不同的計(jì)算機(jī)之間流動(dòng)，故網(wǎng)絡(luò)帶寬變成了瓶頸，是非常寶貴的，“本地計(jì)算”是最有效的一種節(jié)約網(wǎng)絡(luò)帶寬的手段，業(yè)界把這形容為“移動(dòng)計(jì)算比移動(dòng)數(shù)據(jù)更經(jīng)濟(jì)”。任務(wù)粒度把原始大數(shù)據(jù)集切割成小數(shù)據(jù)集時(shí)，通常讓小數(shù)據(jù)集小于或等于HDFS中一個(gè)Block的大小(缺省是64M)，這樣能夠保證一個(gè)小數(shù)據(jù)集位于一臺(tái)計(jì)算機(jī)上，便于本地計(jì)算。有M個(gè)小數(shù)據(jù)集待處理，就啟動(dòng)M個(gè)Map任務(wù)，注意這M個(gè)Map任務(wù)分布于N臺(tái)計(jì)算機(jī)上并行運(yùn)行，Reduce任務(wù)的數(shù)量R則可由用戶指定。Partition把Map任務(wù)輸出的中間結(jié)果按key的范圍劃分成R份(R是預(yù)先定義的Reduce任務(wù)的個(gè)數(shù))，劃分時(shí)通常使用hash函數(shù)如:hash(key)modR，這樣可以保證某一段范圍內(nèi)的key，一定是由一個(gè)Reduce任務(wù)來(lái)處理，可以簡(jiǎn)化Reduce的過(guò)程。Combine在partition之前，還可以對(duì)中間結(jié)果先做combine，即將中間結(jié)果中有相同key的<key,value>對(duì)合并成一對(duì)。combine的過(guò)程與Reduce的過(guò)程類似，很多情況下就可以直接使用Reduce函數(shù)，但combine是作為Map任務(wù)的一部分，在執(zhí)行完Map函數(shù)后緊接著執(zhí)行的。Combine能夠減少中間結(jié)果中<key,value>對(duì)的數(shù)目，從而減少網(wǎng)絡(luò)流量。Reduce任務(wù)從Map任務(wù)結(jié)點(diǎn)取中間結(jié)果Map任務(wù)的中間結(jié)果在做完Combine和Partition之后，以文件形式存于本地磁盤。中間結(jié)果文件的位置會(huì)通知主控JobTracker,JobTracker再通知Reduce任務(wù)到哪一個(gè)DataNode上去取中間結(jié)果。注意所有的Map任務(wù)產(chǎn)生中間結(jié)果均按其Key用同一個(gè)Hash函數(shù)劃分成了R份，R個(gè)Reduce任務(wù)各自負(fù)責(zé)一段Key區(qū)間。每個(gè)Reduce需要向許多個(gè)Map任務(wù)結(jié)點(diǎn)取得落在其負(fù)責(zé)的Key區(qū)間內(nèi)的中間結(jié)果，然后執(zhí)行Reduce函數(shù)，形成一個(gè)最終的結(jié)果文件。任務(wù)管道有R個(gè)Reduce任務(wù)，就會(huì)有R個(gè)最終結(jié)果，很多情況下這R個(gè)最終結(jié)果并不需要合并成一個(gè)最終結(jié)果。因?yàn)檫@R個(gè)最終結(jié)果又可以做為另一個(gè)計(jì)算任務(wù)的輸入，開(kāi)始另一個(gè)并行計(jì)算任務(wù)。執(zhí)行過(guò)程如下所示：SHIYANJUN@cbbd2ce9428e48b/cygdrive/g/hadoop-0.16.4

$bin/hadoopjarhadoop-0.16.4-examples.jarwordcountinput-diroutput-dir

cygpath:cannotcreateshortnameofg:\hadoop-0.16.4\logs

08/09/1220:42:50INFOjvm.JvmMetrics:InitializingJVMMetricswithprocessName

=JobTracker,sessionId=

08/09/1220:42:51INFOmapred.FileInputFormat:Totalinputpathstoprocess:3

08/09/1220:42:51INFOmapred.JobClient:Runningjob:job_local_1

08/09/1220:42:51INFOmapred.MapTask:numReduceTasks:1

08/09/1220:42:52INFOmapred.JobClient:map0%reduce0%

08/09/1220:42:52INFOmapred.LocalJobRunner:file:/g:/hadoop-0.16.4/input-dir/i

nput-a.txt:0+57

08/09/1220:42:52INFOmapred.TaskRunner:Task'job_local_1_map_0000'done.

08/09/1220:42:52INFOmapred.TaskRunner:Savedoutputoftask'job_local_1_map_

0000'tofile:/g:/hadoop-0.16.4/output-dir

08/09/1220:42:53INFOmapred.MapTask:numReduceTasks:1

08/09/1220:42:53INFOmapred.LocalJobRunner:file:/g:/hadoop-0.16.4/input-dir/i

nput-b.txt:0+48

08/09/1220:42:53INFOmapred.TaskRunner:Task'job_local_1_map_0001'done.

08/09/1220:42:53INFOmapred.TaskRunner:Savedoutputoftask'job_local_1_map_

0001'tofile:/g:/hadoop-0.16.4/output-dir

08/09/1220:42:53INFOmapred.MapTask:numReduceTasks:1

08/09/1220:42:53INFOmapred.LocalJobRunner:file:/g:/hadoop-0.16.4/input-dir/i

nput-c.txt:0+56

08/09/1220:42:53INFOmapred.TaskRunner:Task'job_local_1_map_0002'done.

08/09/1220:42:53INFOmapred.TaskRunner:Savedoutputoftask'job_local_1_map_

0002'tofile:/g:/hadoop-0.16.4/output-dir

08/09/1220:42:53INFOmapred.JobClient:map100%reduce0%

08/09/1220:42:54INFOmapred.LocalJobRunner:reduce>reduce

08/09/1220:42:54INFOmapred.TaskRunner:Task'reduce_z7f1uq'done.

08/09/1220:42:54INFOmapred.TaskRunner:Savedoutputoftask'reduce_z7f1uq't

ofile:/g:/hadoop-0.16.4/output-dir

08/09/1220:42:54INFOmapred.JobClient:Jobcomplete:job_local_1

08/09/1220:42:54INFOmapred.JobClient:Counters:9

08/09/1220:42:54INFOmapred.JobClient:

Map-ReduceFramework

08/09/1220:42:54INFOmapred.JobClient:

Mapinputrecords=3

08/09/1220:42:54INFOmapred.JobClient:

Mapoutputrecords=30

08/09/1220:42:54INFOmapred.JobClient:

Mapinputbytes=161

08/09/1220:42:54INFOmapred.JobClient:

Mapoutputbytes=284

08/09/1220:42:54INFOmapred.JobClient:

Combineinputrecords=30

08/09/1220:42:54INFOmapred.JobClient:

Combineoutputrecords=16

08/09/1220:42:54INFOmapred.JobClient:

Reduceinputgroups=16

08/09/1220:42:54INFOmapred.JobClient:

Reduceinputrecords=16

$上面是根據(jù)給定的輸入的數(shù)據(jù)進(jìn)行執(zhí)行的一個(gè)過(guò)程，其實(shí)結(jié)果已經(jīng)輸出到output-dir目錄中，你可以進(jìn)行查看，在目錄G:\hadoop-0.16.4\output-dir下面生成了兩個(gè)文件：.part-00000.crc和part-00000。查看處理結(jié)果，如下所示：或者，直接到G:\hadoop-0.16.4\output-dir目錄下面打開(kāi)part-00000文件查看即可，內(nèi)容如下所示：actor1

add2

after2

apache1

append1

as6

background1

be2

believe1

bench3

block1

cafe2

cat4

communications1

connection1

cust1同上面的是一樣的。這是一個(gè)非常簡(jiǎn)單的例子，而且Hadoop在其中實(shí)現(xiàn)了Google的MapReduce算法，用以處理數(shù)據(jù)。我們可以簡(jiǎn)單看一下關(guān)于WordCount類的實(shí)現(xiàn)，在Hadoop的發(fā)行包中也附帶了例子的源代碼，WordCount.java類實(shí)現(xiàn)如下所示：packageorg.apache.hadoop.examples;importjava.io.IOException;

importjava.util.ArrayList;

importjava.util.Iterator;

importjava.util.List;

importjava.util.StringTokenizer;importorg.apache.hadoop.conf.Configuration;

importorg.apache.hadoop.conf.Configured;

importorg.apache.hadoop.fs.Path;

importorg.apache.hadoop.io.IntWritable;

importorg.apache.hadoop.io.LongWritable;

importorg.apache.hadoop.io.Text;

importorg.apache.hadoop.mapred.JobClient;

importorg.apache.hadoop.mapred.JobConf;

importorg.apache.hadoop.mapred.MapReduceBase;

importorg.apache.hadoop.mapred.Mapper;

importorg.apache.hadoop.mapred.OutputCollector;

importorg.apache.hadoop.mapred.Reducer;

importorg.apache.hadoop.mapred.Reporter;

importorg.apache.hadoop.util.Tool;

importorg.apache.hadoop.util.ToolRunner;/**/**

*這是一個(gè)HadoopMap/Reduce應(yīng)用的例子。

*讀取輸入的文件，實(shí)現(xiàn)將文件的每一行分解成單個(gè)的單詞并統(tǒng)計(jì)單詞的出現(xiàn)頻率。

*輸出結(jié)果是被分解的單詞的列表及其詞頻。

*使用如下命令可以運(yùn)行：bin/hadoopjarbuild/hadoop-examples.jarwordcount

*

[-m<i>maps</i>][-r<i>reduces</i>]<i>in-dir</i><i>out-dir</i>

*/

publicclassWordCountextendsConfiguredimplementsTool{

/**

*MapClass是一個(gè)內(nèi)部靜態(tài)類。統(tǒng)計(jì)數(shù)據(jù)文件中每一行的單詞。

*/

publicstaticclassMapClassextendsMapReduceBase

implementsMapper<LongWritable,Text,Text,IntWritable>{

privatefinalstaticIntWritableone=newIntWritable(1);

privateTextword=newText();

publicvoidmap(LongWritablekey,Textvalue,

OutputCollector<Text,IntWritable>output,

Reporterreporter)throwsIOException{

Stringline=value.toString();

StringTokenizeritr=newStringTokenizer(line);

while(itr.hasMoreTokens()){

word.set(itr.nextToken());

output.collect(word,one);

}

}

}

/**

*Reduce是一個(gè)內(nèi)部靜態(tài)類。作為統(tǒng)計(jì)單詞數(shù)量的中間結(jié)果類，由于這個(gè)例子簡(jiǎn)單無(wú)須執(zhí)行中間結(jié)果的合并。

*/

publicstaticclassReduceextendsMapReduceBase

implementsReducer<Text,IntWritable,Text,IntWritable>{

publicvoidreduce(Textkey,Iterator<IntWritable>values,

OutputCollector<Text,IntWritable>output,

Reporterreporter)throwsIOException{

intsum=0;

while(values.hasNext()){

sum+=values.next().get();

}

output.collect(key,newIntWritable(sum));

}

}

staticintprintUsage(){//提示輸入命令的用法

System.out.println("wordcount[-m<maps>][-r<reduces>]<input><output>");

ToolRunner.printGenericCommandUsage(System.out);

return-1;

}

/**

*map/reduce程序的驅(qū)動(dòng)部分，用于實(shí)現(xiàn)提交map/reduce任務(wù)。

*/

publicintrun(String[]args)throwsException{

JobConfconf=newJobConf(getConf(),WordCount.class);

conf.setJobName("wordcount");

//thekeysarewords(strings)

conf.setOutputKeyClass(Text.class);

//thevaluesarecounts(ints)

conf.setOutputValueClass(IntWritable.class);

conf.setMapperClass(MapClass.class);

conf.setCombinerClass(Reduce.class);

conf.setReducerClass(Reduce.class);

List<String>other_args=newArrayList<String>();

for(inti=0;i<args.length;++i){

try{

if("-m".equals(args[i])){

conf.setNumMapTasks(Integer.parseInt(args[++i]));

}elseif("-r".equals(arg