MapReduce基础案例01-单词计数v1
在本教程中，我们带领大家实现一个“单词记数”的MapReduce编程案例，学习和掌握MapReduce编程和MapReduce数据处理过程。
案例描述
单词记数问题(Word Count)：给定一个巨大的文本（如1TB），如何计算单词出现的数目？
使用MapReduce求解该问题
使用MapReduce实现对单词计数这个问题进行计算的步骤如下：
Step 1: 自动对文本进行分割。
Step 2:在分割之后的每一对
Step 3:对输出的结果集归拢、排序(系统自动完成）：
Step 4:通过Reduce操作生成最后结果：
一、创建Java Maven项目
Maven依赖：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>HadoopDemo</groupId>
<artifactId>com.xueai8</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<!--hadoop依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>3.3.1</version>
</dependency>
<!--hdfs文件系统依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>3.3.1</version>
</dependency>
<!--MapReduce相关的依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>3.3.1</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-jobclient</artifactId>
<version>3.3.1</version>
</dependency>
<!--junit依赖-->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<!--编译器插件用于编译拓扑-->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<!--指定maven编译的jdk版本和字符集,如果不指定,maven3默认用jdk 1.5 maven2默认用jdk1.3-->
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source> <!-- 源代码使用的JDK版本 -->
<target>1.8</target> <!-- 需要生成的目标class文件的编译版本 -->
<encoding>UTF-8</encoding><!-- 字符集编码 -->
</configuration>
</plugin>
</plugins>
</build>
</project>
WcMapper.java：
mapper继承自org.apache.hadoop.mapreduce.Mapper接口。 当Hadoop运行时，它接收输入文件中每个新的行作为一个输入交给该mapper。 map函数拆分该行，并且对于每个拆分出来的单词（word）导出（word,1）作为输出。
package com.xueai8.wordcount01;
import java.io.IOException;
import java.util.regex.Pattern;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
/**
*
* 单词记数 - v1.0
*
* Mapper类
*/
public class WcMapper extends Mapper<LongWritable,Text,Text,IntWritable>{
// 定义可重用的key和value对象
private final static Text keyWord = new Text();
private final static IntWritable one = new IntWritable(1);
// 创建一个正则表达式模式(pattern)用来解析输入的每行文本，基于单词边界("\b")
// 单词的边界包括空格、制表位(Tab)和标点符号
private static final Pattern WORD_BOUNDARY = Pattern.compile("\\s*\\b\\s*");
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 将Text对象转为java String类型
String line = value.toString();
// 使用正则表达式模式将一行拆分为单个的单词，基于单词边界进行拆分
// 如果word对象是空的(比如，由空白字符组成)，则继续下一个被解析的对象
for(String word : WORD_BOUNDARY.split(line)){
if(word.isEmpty()){
continue;
}
keyWord.set(word);
context.write(keyWord, one); // 写出 <word, 1>
}
}
}
WcReducer.java：
reduce函数接收所有的值－具有相同的key－作为输入，并输出该key以及该key出现的数量，作为输出输入数据为<单词, [1,1,1,1,1]>。
package com.xueai8.wordcount01;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
/**
*
* 单词记数 - v1.0
*
* Reducer类
*/
public class WcReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
// 定义可重用的value对象
private final static IntWritable count = new IntWritable(1);
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
// 遍历同一个单词的计数，进行累加
int sum = 0;
for(IntWritable v : values){
sum += v.get();
}
count.set(sum);
context.write(key, count);
}
}
WcDriver.java：
作为输入，这个应用程序可以接收任何文本文件。可直接从IDE运行WcDriver类并传递input和output作为参数。
package com.xueai8.wordcount01;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
/**
*
* 单词记数 - v1.0
*
* 驱动程序类
*/
public class WcDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
// 创建配置对象
Configuration conf = new Configuration();
// 解析命令行传入的参数，并判断参数数量是否正确
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length < 2) {
System.out.println("语法：hadoop jar 包名 com.xueai8.wordcount01.WcDriver <input> <output>");
System.exit(-1);
}
// 创建一个Job对象的新实例，并将job对象命名为"wordcount"
Job job = Job.getInstance(conf,"word-count");
// 基于当前的class，设置要用到的jar
job.setJarByClass(WcDriver.class);
// 为程序设置输入和输出路径。通常我们存储输入文件在HDFS上
// 然后在运行时将输入和输出路径作为命令行参数传进来
FileInputFormat.setInputPaths(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
// 为job设置mapper和reducer类
job.setMapperClass(WcMapper.class);
job.setReducerClass(WcReducer.class);
// 指定job输出的key为Text类型，输出的value为IntWritable类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
// 启动该job并等待job结束。使用方法waitForCompletion(boolean verbose)
// 当参数verbose为true时，该方法会报告其mapper和reducer类运行的进度
// 当参数verbose为false时，该方法会报告进度，但不包括mapper和reducer进度
// 在Unix中，0表示程序执行成功，而非0表示执行故障
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
二、配置log4j
在src/main/resources目录下新增log4j的配置文件log4j.properties，内容如下：
log4j.rootLogger = info,stdout
### 输出信息到控制抬 ###
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n
最终完成的项目结构如下：
三、项目打包
打开IDEA下方的终端窗口terminal，执行"mvn clean package"打包命令，如下图所示：
如果一切正常，会提示打jar包成功。如下图所示：
这时查看项目结构，会看到多了一个target目录，打好的jar包就位于此目录下。如下图所示：
四、项目部署
请按以下步骤执行。
1、启动HDFS集群和YARN集群。在Linux终端窗口中，执行如下的脚本：
$ start-dfs.sh $ start-yarn.sh
查看进程是否启动，集群运行是否正常。在Linux终端窗口中，执行如下的命令：
$ jps
这时应该能看到有如下5个进程正在运行，说明集群运行正常：
5542 NodeManager
5191 SecondaryNameNode
4857 NameNode
5418 ResourceManager
4975 DataNode
2、先在本地创建一个输入数据文件word.txt，并编辑内容如下：
Hello World Bye World Hello Hadoop Bye Hadoop Bye Hadoop Hello Hadoop
3、将数据文件word.txt上传到HDFS的/data/mr/目录下。
$ hdfs dfs -mkdir -p /data/mr $ hdfs dfs -put word.txt /data/mr/ $ hdfs dfs -ls /data/mr/
4、提交作业到Hadoop集群上运行。（如果jar包在Windows下，请先拷贝到Linux中。）
在终端窗口中，执行如下的作业提交命令：
$ hadoop jar HadoopDemo-1.0-SNAPSHOT.jar com.xueai8.wordcount01.WcDriver /data/mr /data/mr-output
执行过程如下图所示：
5、查看输出结果。
在终端窗口中，执行如下的HDFS命令，查看输出结果：
$ hdfs dfs -ls /data/mr-output
$ hdfs dfs -cat /data/mr-output/part-r-00000
也可以通过浏览器查看HDFS上的文件系统和输出结果：
