MapReduce基础案例03-单词计数v3
在2.0版本的“单词计数”程序中，已经解决了在统计时区分大小写的问题，但是另一个问题仍然存在：
- 标点符号也会被当作单词而进行统计。
上述问题的解决方法：
- 过滤这些标点符号或助词(A,An,And等)。可以通过创建一个包含被禁止统计单词的list列表，然后程序在运行时跳过它们（不予统计）。
- 在本例中，创建一个分布式缓存文件，让停止词对集群中所有的机器都可用。
- 在实践中，我们更多地是在Java类中创建一个String[]变量来存储被禁单词和标点符号，然后使用。
重构代码要点：
- 在驱动程序提交作业时，将共享的数据文件（在本例中是停止词文件stop_words.txt）发布到分布式缓存中。
- 在mapper程序中，从分布式缓存中读取共享的数据文件（本例中是停止词文件stop_words.txt），以便词频统计时使用。
一、创建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：
主要是重构这个类，从配置参数中获得传入的控制参数，以决定是否过滤停止词。
如果发现要求过滤停止词，则从分布式缓存中读取共享的stop_words.txt内容，存储到HashSet中备用。
package com.xueai8.wordcount03;
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;
import org.apache.hadoop.util.StringUtils;
import org.apache.log4j.Logger;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.HashSet;
import java.util.Set;
import java.util.regex.Pattern;
/**
*
* 单词记数 - v3.0
* v2.0的问题：
* 1)(已解决) 区分大小写，同一单词，大小写不同，会被统计为不同的单词;
* 2) 标点符号也会被当作单词而进行统计
*
*/
public class WcMapper extends Mapper<LongWritable,Text,Text,IntWritable>{
private static final Logger LOG = Logger.getLogger(WcMapper.class);
// 定义可重用的key和value对象
private final static Text keyWord = new Text();
private final static IntWritable one = new IntWritable(1);
// 默认设为单词记数是不区分大小写的
private boolean caseSensitive = false;
/*** 一个HashSet数据结构,用来存储停止词和标点符号 ***/
private Set<String> patternsToSkip = new HashSet<String>();
// 创建一个正则表达式模式(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(Context context) throws IOException, InterruptedException {
Configuration config = context.getConfiguration();
this.caseSensitive = config.getBoolean("wordcount.case.sensitive", false); // 如果此参数无值，则默认值为false
/*** 如果作业在运行时指定了 wordcount.skip.patterns 参数 ***/
if(config.getBoolean("wordcount.skip.patterns", false)){
URI[] localPaths = context.getCacheFiles(); // 从context中获取缓存的文件路径
parseSkipFile(localPaths[0]); // 从指定的缓存文件中读入被禁单词和标点符号
}
}
/*** 自定义方法，用来读取分布式缓存中的停止词文件 ***/
private void parseSkipFile(URI patternsURI) {
LOG.info("读取分布式缓存中的停止词文件: " + patternsURI);
try {
BufferedReader fis = new BufferedReader(new FileReader(new File(patternsURI.getPath()).getName()));
String pattern;
// 从stop_word.txt文件中一行一行读取被禁单词和标点符号，加入到HashSet中
while ((pattern = fis.readLine()) != null) {
patternsToSkip.add(pattern);
}
} catch (IOException ioe) {
System.err.println("当解析缓存文件时出现异常:'" + patternsURI + "' : "
+ StringUtils.stringifyException(ioe));
}
}
@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)){
// 如果当前的这个word是空的，或者是属于停止词，则过滤掉
if (word.isEmpty() || patternsToSkip.contains(word)) {
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.wordcount03;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
/**
*
* 单词记数 - v3.0
*
* 与v1.0/v2.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：
重构这个类。根据作业提交时指定的参数，以决定是否跳过停止词统计。 如果需要跳过停止词统计，需要将停止词数据文件stop_words.txt添加到分布式缓存中。
package com.xueai8.wordcount03;
import com.xueai8.wordcount01.WcReducer;
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;
import org.apache.log4j.Logger;
import java.io.IOException;
/**
*
* 单词记数 - v3.0
*
* 驱动程序类。
*
*/
public class WcDriver {
private static final Logger LOG = Logger.getLogger(WcDriver.class);
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
// 创建配置对象
Configuration conf = new Configuration();
// GenericOptionsParser是Hadoop框架中解析命令行参数的基本类。
// 它能够辨别一些标准的命令行参数，并且将它们配置到参数文件中去。
// 而函数getRemainingArgs()就是获取了剩余的参数，并且将它们组合为数组 otherArgs。
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
// 解析命令行传入的参数，并判断参数数量是否正确
if (otherArgs.length < 2) {
System.out.println("语法：hadoop jar 包名 com.xueai8.wordcount03.WcDriver <input> <output>");
System.exit(-1);
}
// 创建一个Job对象的新实例，并将job对象命名为"word-count3"
Job job = Job.getInstance(conf,"word-count3");
/*** 遍历命令行参数，看看有没有“-skip”这个参数 ***/
for (int i = 0; i < args.length; i += 1) {
if ("-skip".equals(args[i])) {
// 在配置对象中添加一个boolean类型的变量wordcount.skip.patterns,值设为true
job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
i += 1;
// 向job添加缓存文件路径
job.addCacheFile(new Path(args[i]).toUri());
// 日志记录
LOG.info("将文件添加到分布式缓存: " + args[i]);
break;
}
}
// 基于当前的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.setCombinerClass(WcReducer.class); // 设置combiner
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
另外再创建一个停止词数据文件stop_words.txt，并编辑内容如下：
a an and but is a the to . , !
3、将数据文件word.txt和停止词文件stop_words.txt分别上传到HDFS的/data/mr/目录下。
$ hdfs dfs -mkdir -p /data/mr $ hdfs dfs -put word.txt /data/mr/ $ hdfs dfs -put stop_words.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.wordcount03.WcDriver -Dwordcount.case.sensitive=true /data/mr /data/mr-output -skip stop_words.txt
5、查看输出结果。
在终端窗口中，执行如下的HDFS命令，查看输出结果：
$ hdfs dfs -ls /data/mr-output
$ hdfs dfs -cat /data/mr-output/part-r-00000
6、再一次提交作业并执行（注意，HDFS输出路径要事先不存在，所以要先删除/data/mr-output目录），不指定 -skip 选项参数。然后观察输出结果，看有什么不同。
