可以通過一個(gè)簡單的例子來說明MapReduce到底是什么:
創(chuàng)新互聯(lián)公司專注為客戶提供全方位的互聯(lián)網(wǎng)綜合服務(wù),包含不限于成都做網(wǎng)站、成都網(wǎng)站建設(shè)、湖里網(wǎng)絡(luò)推廣、微信小程序開發(fā)、湖里網(wǎng)絡(luò)營銷、湖里企業(yè)策劃、湖里品牌公關(guān)、搜索引擎seo、人物專訪、企業(yè)宣傳片、企業(yè)代運(yùn)營等,從售前售中售后,我們都將竭誠為您服務(wù),您的肯定,是我們最大的嘉獎(jiǎng);創(chuàng)新互聯(lián)公司為所有大學(xué)生創(chuàng)業(yè)者提供湖里建站搭建服務(wù),24小時(shí)服務(wù)熱線:028-86922220,官方網(wǎng)址:www.cdcxhl.com
我們要統(tǒng)計(jì)一個(gè)大文件中的各個(gè)單詞出現(xiàn)的次數(shù)。由于文件太大。我們把這個(gè)文件切分成如果小文件,然后安排多個(gè)人去統(tǒng)計(jì)。這個(gè)過程就是”Map”。然后把每個(gè)人統(tǒng)計(jì)的數(shù)字合并起來,這個(gè)就是“Reduce"。
上面的例子如果在MapReduce去做呢,就需要?jiǎng)?chuàng)建一個(gè)任務(wù)job,由job把文件切分成若干獨(dú)立的數(shù)據(jù)塊,并分布在不同的機(jī)器節(jié)點(diǎn)中。然后通過分散在不同節(jié)點(diǎn)中的Map任務(wù)以完全并行的方式進(jìn)行處理。MapReduce會對Map的輸出地行收集,再將結(jié)果輸出送給Reduce進(jìn)行下一步的處理。
對于一個(gè)任務(wù)的具體執(zhí)行過程,會有一個(gè)名為"JobTracker"的進(jìn)程負(fù)責(zé)協(xié)調(diào)MapReduce執(zhí)行過程中的所有任務(wù)。若干條TaskTracker進(jìn)程用來運(yùn)行單獨(dú)的Map任務(wù),并隨時(shí)將任務(wù)的執(zhí)行情況匯報(bào)給JobTracker。如果一個(gè)TaskTracker匯報(bào)任務(wù)失敗或者長時(shí)間未對本身任務(wù)進(jìn)行匯報(bào),JobTracker會啟動另外一個(gè)TaskTracker重新執(zhí)行單獨(dú)的Map任務(wù)。
(1)eclipse下創(chuàng)建相關(guān)maven項(xiàng)目,依賴jar包如下(也可參照hadoop源碼包下的hadoop-mapreduce-examples項(xiàng)目的pom配置)
注意:要配置一個(gè)maven插件maven-jar-plugin,并指定mainClass
junit junit 4.11 org.apache.hadoop hadoop-mapreduce-client-core 2.5.2 org.apache.hadoop hadoop-common 2.5.2 org.apache.maven.plugins maven-jar-plugin com.xxx.demo.hadoop.wordcount.WordCount
(2)根據(jù)MapReduce的運(yùn)行機(jī)制,一個(gè)job至少要編寫三個(gè)類分別用來完成Map邏輯、Reduce邏輯、作業(yè)調(diào)度這三件事。
Map的代碼可繼承org.apache.hadoop.mapreduce.Mapper類
public static class TokenizerMapper extends Mapper
Reduce的代碼可繼承org.apache.hadoop.mapreduce.Reducer類
public class IntSumReducer extends Reducer{ private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } }
編寫main方法進(jìn)行作業(yè)調(diào)度
public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); job.waitForCompletion(true) ; //System.exit(job.waitForCompletion(true) ? 0 : 1); }
執(zhí)行mvn install把項(xiàng)目打成jar文件然后上傳到linux集群環(huán)境,使用hdfs dfs -mkdir命令在hdfs文件系統(tǒng)中創(chuàng)建相應(yīng)的命令,使用hdfs dfs -put 把需要處理的數(shù)據(jù)文件上傳到hdfs系統(tǒng)中,示例:hdfs dfs -put ${linux_path/數(shù)據(jù)文件} ${hdfs_path}
在集群環(huán)境中執(zhí)行命令: hadoop jar ${linux_path}/wordcount.jar ${hdfs_input_path} ${hdfs_output_path}
hdfs dfs -cat ${hdfs_output_path}/輸出文件名
以上的方式在未啟動hadoop集群環(huán)境時(shí),是以Local模式運(yùn)行,此時(shí)HDFS和YARN都不起作用。下面是在偽分布式模式下執(zhí)行mapreduce job時(shí)需要做的工作,先把官網(wǎng)上列的步驟摘錄出來:
配置主機(jī)名
# vi /etc/sysconfig/network
例如:
NETWORKING=yes HOSTNAME=master vi /etc/hosts
填入以下內(nèi)容
127.0.0.1 localhost
配置ssh免密碼互通
ssh-keygen -t rsa
# cat?~/.ssh/id_rsa.pub?>>?~/.ssh/authorized_keys
配置core-site.xml文件(位于${HADOOP_HOME}/etc/hadoop/
fs.defaultFS hdfs://localhost:9000
配置hdfs-site.xml文件
dfs.replication 1
下面的命令可以在單機(jī)偽分布模式下運(yùn)行mapreduce的job
1.Format the filesystem:
$ bin/hdfs namenode -format
2.Start NameNode daemon and DataNode daemon:
$ sbin/start-dfs.sh
3.The hadoop daemon log output is written to the $HADOOP_LOG_DIR directory (defaults to $HADOOP_HOME/logs).4.Browse the web interface for the NameNode; by default it is available at:
NameNode - http://localhost:50070/
Make the HDFS directories required to execute MapReduce jobs:
$ bin/hdfs dfs -mkdir /user
$ bin/hdfs dfs -mkdir /user/
5.Copy the input files into the distributed filesystem:
$ bin/hdfs dfs -put etc/hadoop input
6.Run some of the examples provided:
$ bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.5.2.jar grep input output 'dfs[a-z.]+'
7.Examine the output files:
Copy the output files from the distributed filesystem to the local filesystem and examine them:$ bin/hdfs dfs -get output output
$ cat output/*
orView the output files on the distributed filesystem:
$ bin/hdfs dfs -cat output/*
8.When you're done, stop the daemons with:
$ sbin/stop-dfs.sh
以上就是本文關(guān)于hadoop的wordcount實(shí)例代碼的全部內(nèi)容,希望對大家有所幫助。感興趣的朋友可以繼續(xù)參閱本站其他相關(guān)專題,如有不足之處,歡迎留言指出。感謝朋友們對本站的支持!