MapReduce基础案例02-单词计数v2
在1.0版本的“单词计数”程序中，存在着如下的问题：
- 区分大小写。同一单词，大小写不同，会被统计为不同的单词;
- 标点符号也会被当作单词而进行统计。
下面我们重构单词计数程序（v2.0版），解决上面的第一个问题，即在统计时区分大小写问题。
在重构后的v2.0版本单词计数程序中，我们可以在运行时动态控制是否区分单词的大小写。
一、创建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.wordcount02;
import java.io.IOException;
import java.util.regex.Pattern;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
/**
*
* 单词记数 - v2.0
*
* v1.0的问题：
* 1) 区分大小写。同一单词，大小写不同，会被统计为不同的单词;
* 2) 标点符号也会被当作单词而进行统计。
*
* 运行：
* 要区分大小写进行统计的话，在运行时命令行传递一个系统变量：-Dwordcount.case.sensitive=true
*/
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);
/*** 默认设为单词记数是不区分大小写的 ***/
private boolean caseSensitive = false;
// 创建一个正则表达式模式(pattern)用来解析输入的每行文本，基于单词边界("\b")
// 单词的边界包括空格、制表位(Tab)和标点符号
private static final Pattern WORD_BOUNDARY = Pattern.compile("\\s*\\b\\s*");
/*
* 当提交一个job时，Hadoop会自动地调用这个方法。在这个方法中，实例化一个Configuration对象，
* 然后设置实例变量caseSensitive的值为系统变量wordcount.case.sensitive的值(来自命令行)
*/
@Override
protected void setup(Mapper.Context context) throws IOException, InterruptedException {
Configuration config = context.getConfiguration();
this.caseSensitive = config.getBoolean("wordcount.case.sensitive", false); // 如果此参数无值，则默认值为false
}
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 将Text对象转为java String类型
String line = value.toString();
/*** 如果不区分大小写 ***/
if(!caseSensitive){
line = line.toLowerCase(); // 将一行文本全都转换为小写
}
// 使用正则表达式模式将一行拆分为单个的单词，基于单词边界进行拆分
// 如果word对象是空的(比如，由空白字符组成)，则继续下一个被解析的对象
for(String word : WORD_BOUNDARY.split(line)){
if(word.isEmpty()){
continue;
}
keyWord.set(word);
context.write(keyWord, one); // 写出 <word, 1>
}
}
}
WcReducer.java：(保持不变，与v1.0中相同)
reduce函数接收所有的值－具有相同的key－作为输入，并输出该key以及该key出现的数量，作为输出输入数据为<单词, [1,1,1,1,1]>。
package com.xueai8.wordcount02;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
/**
*
* 单词记数 - v2.0
*
* 与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;
/**
*
* 单词记数 - v2.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.worcount02.WcDriver <input> <output>");
System.exit(-1);
}
// 创建一个Job对象的新实例，并将job对象命名为"word-count2"
Job job = Job.getInstance(conf,"word-count2");
// 基于当前的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中。）
在终端窗口中，执行如下的作业提交命令（注意命令行的-Dwordcount.case.sensitive=true参数）：
$ hadoop jar HadoopDemo-1.0-SNAPSHOT.jar com.xueai8.wordcount02.WcDriver -Dwordcount.case.sensitive=true /data/mr /data/mr-output
5、查看输出结果。
在终端窗口中，执行如下的HDFS命令，查看输出结果：
$ hdfs dfs -ls /data/mr-output
$ hdfs dfs -cat /data/mr-output/part-r-00000
注意到，输出结果是区分大小写的。
