（超详细）MapReduce工作原理及基础编程

MapReduce工作原理及基础编程（代码见文章后半部分） 

JunLeon——go big or go home

目录 

MapReduce工作原理及基础编程（代码见文章后半部分）

一、MapReduce概述

1SEO靠我、什么是MapReduce？

2、WordCount案例解析MapReduce计算过程

（1）运行hadoop自带的样例程序

（2）MapReduce工作过程

3、Shuffle过程详解

二、MapReduceSEO靠我编程基础

1、Hadoop数据类型

2、数据输入格式InputFormat

3、输入数据分块InputSplit和数据记录读入RecordReader

4、数据输出格式OutputFormat

5、数据记录输出SEO靠我类RecordWriter

 6、Mapper类

7、Reduce类

三、MapReduce项目案例

1、经典案例——WordCount

2、计算考试平均成绩

3、网站日志分析

 前言：

Google于2003年在SOSSEO靠我P上发表了《The Google File System》，于2004年在OSDI上发表了《MapReduce: Simplified Data Processing on Large ClusterSEO靠我s》，于2006年在OSDI上发表了《Bigtable: A Distributed Storage System for Structured Data》。这三篇论文为大数据及云计算的发展奠定了基础SEO靠我。

一、MapReduce概述

1、什么是MapReduce？

        MapReduce是一个分布式、并行处理的计算框架。

        MapReduce 把任务分为 Map 阶段和 Reduce阶段。开发人员使用存储在HDFS 中SEO靠我数据（可实现快速存储），编写 Hadoop 的 MapReduce 任务。由于 MapReduce工作原理的特性， Hadoop 能以并行的方式访问数据，从而实现快速访问数据。

表1 map函数和rudSEO靠我ece函数

函数输入输出说明map

<k1,v1>

<0,helle world>

<12,hello hadoop>

List<k2,v2>

<hello,1>

<world,1>

<hello,1>

<hhadooSEO靠我p,1>

将获取到的数据集进一步解析成<key,value>,通过Map函数计算生成中间结果，进过shuffle处理后作为reduce的输入reduce

<k2,List(v2)> 

<hadoop,1>

<hSEO靠我ello,{1,1}>

<world,1>

<k3,v3>

<hadoop,1>

<hello,2>

<world,1>

reduce得到map输出的中间结果，合并计算将最终结果输出HDFS，其中List(v2)SEO靠我，指同一k2的value

 MapReduce体系结构主要由四个部分组成，分别是：Client、JobTracker、TaskTracker以及Task

　　1）Client

用户编写的MapReduce程序通过SEO靠我Client提交到JobTracker端 用户可通过Client提供的一些接口查看作业运行状态。

　　2）JobTracker

JobTracker负责资源监控和作业调度 JobTracker 监控所有TasSEO靠我kTracker与Job的健康状况，一旦发现失败，就将相应的任务转移到其他节点 JobTracker 会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器（TaskScheduler），SEO靠我而调度器会在资源出现空闲时，选择合适的任务去使用这些资源。

　　3）TaskTracker

TaskTracker 会周期性地通过“心跳”将本节点上资源的使用情况和任务的运行进度汇报给JobTracker，同SEO靠我时接收JobTracker 发送过来的命令并执行相应的操作（如启动新任务、杀死任务等） TaskTracker 使用“slot”等量划分本节点上的资源量（CPU、内存等）。一个Task 获取到一个slSEO靠我ot 后才有机会运行，而Hadoop调度器的作用就是将各个TaskTracker上的空闲slot分配给Task使用。slot 分为Map slot 和Reduce slot 两种，分别供MapTaskSEO靠我 和Reduce Task 使用。

　　4）Task

　　Task 分为Map Task 和Reduce Task 两种，均由TaskTracker 启动。

MapReduce各个执行阶段：

MapReduce应用程SEO靠我序执行过程：

可以参考大佬黎先生的博客：MapReduce基本原理及应用 - 黎先生 - 博客园

2、WordCount案例解析MapReduce计算过程

（1）运行hadoop自带的样例程序

WordCouSEO靠我nt案例是一个经典案例，是Hadoop自带的样例程序。

作用：统计单词数量（出现的次数）

应用：求和、求平均值、求最值，

jar包存储在$HADOOP_HOME/share/hadoop/mapreduceSEO靠我/：

$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.3.jar

例如：

 步骤：

1.在本地创建一个文件

输入以下内容：

2.上SEO靠我传到HDFS指定目录

在HDFS中创建指定文件：

 上传文件：

 3.使用hadoop jar命令运行jar程序，统计单词数量

 4.输出结果

执行部分过程：

 查看生成的文件：

 查看计算结果：

（2）MapReduce工作SEO靠我过程

        工作流程是Input从HDFS里面并行读取文本中的内容，经过MapReduce模型，最终把分析出来的结果用Output封装，持久化到HDFS中。

1.Mapper工作过程：

 附上Mapper阶段代码：SEO靠我

public static class WorldCount_Mapper extends Mapper<LongWritable, Text, Text, IntWritable>{@OverridSEO靠我eprotected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.ContSEO靠我ext context)throws IOException, InterruptedException {System.out.println("split:<" + key + ","+ valuSEO靠我e + ">" );String[] strs = value.toString().split(" ");for (String string : strs) {System.out.printlnSEO靠我("map:<" + key + ","+ value + ">" );context.write(new Text(string),new IntWritable(1));}}}

KEYIN--LonSEO靠我gWritable:输入key类型，记录数据分片的偏移位置

VALUEIN—Text:输入的value类型，对应分片中的文本数据

KEYOUT--Text:输出的key类型，对应map方法中计算结果的keSEO靠我y值

VALUEOUT—IntWritable:输出的value类型，对应map方法中计算结果的value值

Mapper类从分片后传出的上下文中接收数据，数据以类型<LongWritable,Text>SEO靠我的键值对接收过来，通过重写map方法默认一行一行的读取数据并且以<key,value>形式进行遍历赋值。

2.Reducer工作过程：

 附上Reducer阶段代码:

public static class SEO靠我WorldCount_Reducer extends Reducer<Text, IntWritable, Text, IntWritable>{@Overrideprotected void redSEO靠我uce(Text key, Iterable<IntWritable> values,Reducer<Text, IntWritable, Text, IntWritable>.Context conSEO靠我text) throws IOException, InterruptedException {int index = 0;for (IntWritable intWritable : values)SEO靠我 {System.out.println("reduce:<" + key + ","+ intWritable + ">" );index += intWritable.get();}contextSEO靠我.write(key,new IntWritable(index));}}

Reducer任务继承Reducer类，主要接收的数据来自Map任务的输出，中间经过Shuffle分区、排序、分组，最终以<kSEO靠我ey,value>形式输出给用户。

Job提交代码：

public static void main(String[] args) throws IOException, ClassNotFoundExcSEO靠我eption, InterruptedException {Job job = Job.getInstance();job.setJarByClass(WorldCount.class);job.seSEO靠我tOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);job.setMapperClass(WorldCountSEO靠我_Mapper.class);job.setReducerClass(WorldCount_Reducer.class);FileInputFormat.addInputPath(job,new PaSEO靠我th("hdfs://192.168.100.123:8020/input"));FileOutputFormat.setOutputPath(job, new Path("hdfs://192.16SEO靠我8.100.123:8020/output"));job.waitForCompletion(true);}

JobClients是用户提交的作业与ResourceManager交互的主要接口，JobCSEO靠我lients提供提交作业、追踪进程、访问子任务的日志记录、获取的MapReduce集群状态信息等功能。

3、Shuffle过程详解

Hadoop运行机制中，将map输出进行分区、分组、排序、和合并等处理后SEO靠我作为输入传给Reducer的过程，称为shuffle过程。

 shuffle阶段又可以分为Map端的shuffle和Reduce端的shuffle。

　　一、Map端的shuffle

　　写磁盘：Map端会处理输入数SEO靠我据并产生中间结果，这个中间结果会写到本地磁盘，而不是HDFS。每个Map的输出会先写到内存缓冲区中，当写入的数据达到设定的阈值时，系统将会启动一个线程将缓冲区的数据写到磁盘，这个过程叫做spill。

　　分SEO靠我区、分组、排序：在spill写入之前，会先进行二次排序，首先根据数据所属的partition进行排序，然后每个分区（partition）中的数据再按key来排序。partition的目是将记录划分到不SEO靠我同的Reducer上去，以期望能够达到负载均衡，以后的Reducer就会根据partition来读取自己对应的数据。接着运行combiner(如果设置了的话)，combiner的本质也是一个ReducSEO靠我er，其目的是对将要写入到磁盘上的文件先进行一次处理，这样，写入到磁盘的数据量就会减少。最后将数据写到本地磁盘产生spill文件(spill文件保存在{mapred.local.dir}指定的目录中，SEO靠我Map任务结束后就会被删除)。

        文件合并：最后，每个Map任务可能产生多个溢写文件（spill file），在每个Map任务完成前，会通过多路归并算法将这些spill文件归并成一个已经分区和排序的输出文SEO靠我件。至此，Map的shuffle过程就结束了。

        压缩：在shuffle过程中如果压缩被启用，在map传出数据传入Reduce之前可执行压缩，默认情况下压缩是关闭的，可以将mapred.compress.SEO靠我map.output设置为true可实现压缩。

　　二、Reduce端的shuffle

　　Reduce端的shuffle主要包括三个阶段，copy、sort(merge)和reduce。

首先要将Map端产生的输SEO靠我出文件拷贝到Reduce端，但每个Reducer如何知道自己应该处理哪些数据呢？因为Map端进行partition的时候，实际上就相当于指定了每个Reducer要处理的数据(partition就对应了SEO靠我Reducer)，所以Reducer在拷贝数据的时候只需拷贝与自己对应的partition中的数据即可。每个Reducer会处理一个或者多个partition，但需要先将自己对应的partition中SEO靠我的数据从每个Map的输出结果中拷贝过来。

接下来就是排序（sort）阶段，也成为合并（merge）阶段，因为这个阶段的主要工作是执行了归并排序。从Map端拷贝到Reduce端的数据都是有序的，所以很适合SEO靠我归并排序。最终在Reduce端生成一个较大的文件作为Reduce的输入。MapReduce编程接口

二、MapReduce编程基础

1、Hadoop数据类型

Hadoop数据包括：BooleanWritabSEO靠我le、ByteWritable、DoubleWritable、FloatWritale、IntWritable、LongWritable、Text、NullWritable等，它们实现了WritablSEO靠我eComparable接口。其中Text表示使用UTF8格式存储的文本、NullWritable类型是当（key,value）中的key或value为空时使用。

表2 Hadoop Writable与JSEO靠我ava数据类型参照表

Java基本类型Writable封装类类型序列化后的长度为booleanBooleanWritable布尔型1byteByteWritable字节型1doubleDoubleWriSEO靠我table双精度浮点型8floatFloatWritable单精度浮点型8int

IntWritable

VIntWritable

整型

4

1-5

long

LongWritable

长整型8shortShortWSEO靠我ritable短整型2nullNullWritable空值0Text文本类型

除了上述Hadoop类型外，用户还可以自定义新的数据类型。用户自定义数据类型需要实现Writable接口，但如果需要作为主键SEO靠我key使用或需要比较大小时，则需要实现WritableComparable接口。

2、数据输入格式InputFormat

抽象类InputFormat<K,V>有三个直接子类：

FileInputFormaSEO靠我t<K,V>、DBInputFormat<T>、DelegatingInputFormat<K,V>

其中，文件输入格式类FileInputFormat<K,V>类有几个子类：

TextInputFormSEO靠我at、KeyValueInputFormat、SequenceFileInputFormat<K,V>、NlineInputFormat、CombineFileInputFormat<K,V>

序列化文SEO靠我件输入类SequenceFileInputFormat<K,V>有几个子类：

SequenceFileAsBinaryInputFormat、SequenceFileAsTextInputFormat、SEO靠我SequenceFileInputFilter<K,V>

数据库输入格式类DBInputFormat<T>的直接子类是：DataDriverDBInputFormat<T>，而这个子类又派生子类：OraSEO靠我cleDataDriverDBInputFormat<T>

表3  常用数据输入格式类

InputFormat类描述键（Key）值（Value）TextInputFormat默认输入格式，读取文本文件的行SEO靠我当前行的偏移量当前行内容KeyValueTextInputFormat将行解析成键值对行内首个制表符的内容行内其余内容SequenceFileInputFormat专用于高性能的二进制格式用户定义用户SEO靠我定义

3、输入数据分块InputSplit和数据记录读入RecordReader

编程时由用户选择的数据输入格式InputFormat类型来自动决定数据分块InputSplit和数据记录RecordReaSEO靠我der类型。一个InputSplit将单独作为一个Mapper的输入，即作业的Mapper数量是由InputSplit个数决定的。

表4 数据输出格式类对应的Reader类型

InputFormat类ReSEO靠我cordReader类描述TextInputFormatLineRecordReader读取文本文件的行KeyValueTextInputFormatKeyValueLineRecordReader读SEO靠我取行并将行解析为键值对SequenceFileInputFormatSequenceFileRecordReader用户定义的格式产生键与值DBInputFormatDBRecordReader仅适合SEO靠我读取少量数据记录，不适合数据仓库联机数据分析大量数据的读取处理

4、数据输出格式OutputFormat

抽象类OutputFormat<K,V>有四个直接子类：

FileOutputFormat<K,V>SEO靠我、DBOutputFormat<K,V>、NullOutputFormat<K,V>、FilterOutputFormat<K,V>

FileOutputFormat<K,V>有两个直接子类：

TextOSEO靠我utputFormat<K,V>、SequenceFileOutputFormat<K,V>

SequenceFileOutputFormat<K,V>有直接子类：SequenceFileAsBinarSEO靠我yOutputFormat

FilterOutputFormat<K,V>有直接子类：LazyOutputFormat<K,V>

5、数据记录输出类RecordWriter

数据记录输出类RecordWriSEO靠我ter是一个抽象类。

表5 数据输出格式类对应的数据记录Writer类型

OutputFormat类RecordWriter类描述TextOutputFormatLineRecordWriter将结果数据SEO靠我以“key + \t + value”形式输出到文本文件中SequenceFileOutputFormatSequenceFileRecordWriter用户定义的格式产生键与值DBOutputForSEO靠我matDBRecordWriter将结果写入到一个数据库表中FilterOutputFormatFilterRecordWriter对应于过滤器输出模式的数据记录模式，只将过滤器的结果输出到文件中

6、SEO靠我Mapper类

Mapper类是一个抽象类，位于hadoop-mapreduce-client-core-2.x.x.jar中，其完整类名是：org.apache.hadoop.mapreduce.MaSEO靠我pper<KEYIN,VALUEIN,KEYOUT,VALUEOUT>，需派生子类使用，在子类中重写map方法：map(KEYIN key,VALUEIN value,Mapper.Context cSEO靠我ontext)对出入的数据分块每个键值对调用一次。

7、Reduce类

Reduce类是一个抽象类，位于hadoop-mapreduce-client-core-2.x.x.jar中，其完整类名是：orgSEO靠我.apache.hadoop.mapreduce.Reduce<KEYIN,VALUEIN,KEYOUT,VALUEOUT>，需派生子类使用，在子类中重写reduce方法：reduce(KEYIN kSEO靠我ey,Inerable <VALUEIN> value,Reducer.Context context)对出入的数据分块每个键值对调用一次。

三、MapReduce项目案例

1、经典案例——WordCouSEO靠我nt

代码演示：

package hadoop.mapreduce;import java.io.IOException;import org.apache.hadoop.conf.ConfiguratiSEO靠我on; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; SEO靠我 import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; SEO靠我 import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import orSEO靠我g.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; importSEO靠我 org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lSEO靠我ib.output.FileOutputFormat;public class MyWordCount {/** KEYIN：是map阶段输入的key（偏移量）* VALUEIN：是map阶段输入的vSEO靠我alue（文本文件的内容--行）* KEYOUT：是map阶段输出的key(单词)* VALUEOUT：是map阶段输出的value（单词的计数--1）* * Java基本数据类型：* int、shoSEO靠我rt、long、double、float、char、boolean、byte* hadoop数据类型* IntWritable、ShortWritable、LongWritable、DoubleWriSEO靠我table、FloatWritable* ByteWritable、BooleanWritable、NullWritable、Text* Text：使用utf8编码的文本类型*/public statSEO靠我ic class WordCount_Mapper extends Mapper<LongWritable, Text, Text, IntWritable>{@Override //方法的重写proSEO靠我tected void map(LongWritable key, Text value, Mapper<LongWritable, Text,Text, IntWritable>.Context cSEO靠我ontext)throws IOException, InterruptedException {String[] line = value.toString().split(" "); //将获取到SEO靠我的数据以空格进行切分成一个个单词for (String word : line) { //遍历单词的数组context.write(new Text(word), new IntWritable(1)SEO靠我); //单词进行计数，将中间结果写入context}} }/** KEYIN：reduce阶段输入的key(单词)* VALUEIN：reduce阶段输入的value(单词的计数)* KEYOUT：SEO靠我reduce阶段输出的key(单词)* VALUEOUT：reduce阶段输出的value(单词计数的总和)* * reduce方法中做以下修改：* 将Text arg0改为Text key* 将ItSEO靠我erable<IntWritable> arg1改为Iterable<IntWritable> value* 将Context arg2修改为Context context*/public statiSEO靠我c class WordCount_Reducer extends Reducer<Text, IntWritable, Text, IntWritable>{@Overrideprotected vSEO靠我oid reduce(Text key, Iterable<IntWritable> values,Reducer<Text, IntWritable, Text, IntWritable>.ContSEO靠我ext context)throws IOException, InterruptedException {int sum = 0; //创建一个变量,和for (IntWritable intWriSEO靠我table : values) { //遍历相同key单词的计数sum += intWritable.get(); //将相同key单词的计数进行累加}context.write(key, new ISEO靠我ntWritable(sum)); //将计算的结果写入context}}//提交工作public static void main(String[] args) throws Exception {SEO靠我String inPath= "hdfs://192.168.182.10:8020/input.txt";String outPath = "hdfs://192.168.182.10:8020/oSEO靠我utput/";Configuration conf = new Configuration();Job job = Job.getInstance(); //创建Job对象jobFileSystemSEO靠我 fs = FileSystem.get(conf);if (fs.exists(new Path(outPath))) {fs.delete(new Path(outPath), true);}joSEO靠我b.setJarByClass(MyWordCount.class); //设置运行的主类MyWordCountjob.setMapperClass(WordCount_Mapper.class); SEO靠我//设置Mapper的主类job.setReducerClass(WordCount_Reducer.class); //设置Reduce的主类job.setOutputKeyClass(Text.cSEO靠我lass); //设置输出key的类型job.setOutputValueClass(IntWritable.class); //设置输出value的类型//设置文件的输入路径(根据自己的IP和HDFSEO靠我S地址设置)FileInputFormat.addInputPath(job, new Path(inPath)); //设置计算结果的输出路径(根据自己的IP和HDFS地址设置)FileOutputSEO靠我Format.setOutputPath(job, new Path(outPath));System.exit((job.waitForCompletion(true)?0:1)); //提交任务并SEO靠我等待任务完成} }

打包上传虚拟机：

步骤：

右键单击项目名 --> 选择 Export --> Java --> JAR file --> Browse...选择存放路径 --> 文件名

命名为wordcounSEO靠我t.jar，将打包好的jar包上传到虚拟机中

运行代码：

在本地创建一个文件input.txt

vi input.txt

 添加内容：

hello world hello hadoop SEO靠我 bye world bye hadoop

上传到DHFS中：

hadoop fs -put input.txt /

使用jar命令执行项目：

hadoop jar wordcount.jSEO靠我ar hadoop.mapreduce.MyWordCount

如下图：

查看结果： 

2、计算考试平均成绩

代码演示：

Mapper类

package hadoop.mapreduce;import java.iSEO靠我o.IOException; import java.util.StringTokenizer;import org.apache.hadoop.io.IntWritable; SEO靠我 import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; impSEO靠我ort org.apache.hadoop.mapreduce.Counter; import org.apache.hadoop.mapreduce.Mapper;/** 编写CouSEO靠我rseScoreAverageMapper继承Mapper类*/ public class CourseScoreAverageMapper extends Mapper<LongWrSEO靠我itable, Text, Text, IntWritable>{@Override //方法的重写protected void map(LongWritable key, Text value, MSEO靠我apper<LongWritable, Text,Text, IntWritable>.Context context)throws IOException, InterruptedExceptionSEO靠我 {String line = new String(value.getBytes(),0,value.getLength(),"UTF8"); //转换中文编码Counter countPrint SEO靠我= context.getCounter("CourseScoreAverageMapper.Map 输出传递Value:", line); //通过计数器输出变量值countPrint.incremSEO靠我ent(1L); //将计数器加一StringTokenizer tokenArticle = new StringTokenizer(line,"\n"); //将输入的数据按行“\n”进行分割whSEO靠我ile(tokenArticle.hasMoreElements()) {StringTokenizer tokenLine = new StringTokenizer(tokenArticle.neSEO靠我xtToken()); //每行按空格划分String strName = tokenLine.nextToken(); //按空格划分出学生姓名String strScore = tokenLineSEO靠我.nextToken(); //按空格划分出学生成绩Text name = new Text(strName); //转换为Text类型int scoreInt = Integer.parseInt(SEO靠我strScore); //转换为int类型context.write(name, new IntWritable(scoreInt)); //将中间结果写入contextcountPrint = coSEO靠我ntext.getCounter("CourseScoreAverageMapper.Map中循环输出信息：", "<key,value>:<"+strName+","+strScore+">"); SEO靠我//输出信息countPrint.increment(1L); //将计数器加一}} }

Reducer类

package hadoop.mapreduce;import java.io.SEO靠我IOException;import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; SEO靠我 import org.apache.hadoop.mapreduce.Counter; import org.apache.hadoop.mapreduce.Reducer;/*SEO靠我* 编写CourseScoreAverageReducer继承Reduce类*/ public class CourseScoreAverageReducer extends ReduSEO靠我cer<Text, IntWritable, Text, IntWritable>{@Override //重写reduce方法protected void reduce(Text key, IterSEO靠我able<IntWritable> values,Reducer<Text, IntWritable, Text, IntWritable>.Context context)throws IOExceSEO靠我ption, InterruptedException {int sum = 0; //总分int count = 0; //科目数for (IntWritable val : values) { /SEO靠我/遍历相同key的分数sum += val.get(); //将相同key的分数进行累加count++; //计算科目数}int average = (int)sum/count; //计算平均分coSEO靠我ntext.write(key, new IntWritable(average)); //将计算的结果写入contextCounter countPrint = context.getCounterSEO靠我("CourseScoreAverageReducer.Reducer中输出信息：", "<key,value>:<"+key.toString()+","+average+">"); //输出信息cSEO靠我ountPrint.increment(1L); //计数器加1} }

Driver类

package hadoop.mapreduce;import org.apache.hadoop.SEO靠我conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoopSEO靠我.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text;SEO靠我 import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.inpuSEO靠我t.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; impSEO靠我ort org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapredSEO靠我uce.lib.output.TextOutputFormat; import org.apache.hadoop.util.GenericOptionsParser;public cSEO靠我lass CourseScoreDriver {public static void main(String[] args) throws Exception {Configuration conf SEO靠我= new Configuration(); //获取配置文件Job job = Job.getInstance(conf,"CourseScoreAverage"); //创建Job对象jobStrSEO靠我ing[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); //获取命令行参数if(otherArgs.lenSEO靠我gth<2) { System.err.print("Usage:hadoop jar MyAverage.jar <in> <out> ");System.err.print("hadoop jarSEO靠我 MyAverage.jar hadoop.mapreduce.CourseScoreDriver <in> <out>");System.exit(2);}else {for (int i = 0;SEO靠我 i < otherArgs.length-1; i++) { //设置文件输入路径if(!("hadoop.mapreduce.CourseScoreDriver".equalsIgnoreCaseSEO靠我(otherArgs[i]))) { //排除hadoop.mapreduce.CourseScoreDriver这个参数FileInputFormat.addInputPath(job, new PSEO靠我ath(otherArgs[i]));System.out.println("参数IN:"+otherArgs[i]);}}//设置文件输出路径FileOutputFormat.setOutputPaSEO靠我th(job, new Path(otherArgs[otherArgs.length-1])); //设置输出路径System.out.println("参数OUT："+otherArgs[otheSEO靠我rArgs.length-1]);}FileSystem hdfs = FileSystem.get(conf); //创建文件系统if(hdfs.exists(new Path(otherArgs[SEO靠我otherArgs.length-1]))) { //如果已经存在该路径，则删除该路径hdfs.delete(new Path(otherArgs[otherArgs.length-1]), trueSEO靠我);}job.setJarByClass(CourseScoreDriver.class); //设置运行的主类CourseScoreDriverjob.setMapperClass(CourseScSEO靠我oreAverageMapper.class); //设置Mapper的主类job.setCombinerClass(CourseScoreAverageReducer.class); //设置ComSEO靠我biner的主类job.setReducerClass(CourseScoreAverageReducer.class); //设置Reduce的主类job.setOutputKeyClass(TexSEO靠我t.class); //设置输出key的类型job.setOutputValueClass(IntWritable.class); //设置输出value的类型job.setInputFormatClSEO靠我ass(TextInputFormat.class); //设置输入格式job.setOutputFormatClass(TextOutputFormat.class); //设置输出格式SystemSEO靠我.exit((job.waitForCompletion(true)?0:1)); //提交任务并等待任务完成System.out.println("Job Finished!");} SEO靠我 }

打包上传虚拟机：

步骤：

右键单击项目名 --> 选择 Export --> Java --> JAR file --> Browse...选择存放路径 --> 文件名

命名为average.jar ， 将打包好的averSEO靠我age.jar上传到虚拟机中

运行代码：

首先准备三个文件 Chinese.txt、Math.txt、English.txt，添加如下内容：

 将文件上传到HDFS的data目录下： 

hadoop fs -mkSEO靠我dir /data hadoop fs -put Chinese.txt /data/ hadoop fs -put Math.txt /data/ hSEO靠我adoop fs -put English.txt /data/

执行代码：

hadoop jar average.jar hadoop.mapreduce.CourseScoreDriver /dataSEO靠我 /data/output

 查看结果，如下图：

3、网站日志分析

代码演示：

打包上传虚拟机：

运行代码：