初识MapReduce
一、什么是MapReduce
MapReduce是一种编程范式，它借助Map将一个大任务分解成多个小任务，再借助Reduce归并Map的结果。MapReduce虽然原理很简单，但是使用MapReduce设计出一个解决问题的应用却不是一件简单的事情。下面通过一个简单的小例子来介绍MapReduce。
二、使用MapReduce寻找销售人员业绩最大值
《Hadoop权威指南》的例子是寻找天气最大值，需要去下载数据。但是我们并不需要完全复刻他的场景，所以这里用了另外一个例子。假设有一批销售日志数据文件，它的一部分是这样的。
66$2021-01-01$5555
67$2021-01-01$5635
每一行代表某一位销售人员某个日期的销售数量，具体格式为
销售用户id$统计日期$销售数量
我们需要寻找每一个销售用户的销售最大值是多少。需要说明的是，这里仅仅是举一个很简单的示例，便于学习MapReduce。
1、数据解析器
我首先写了一个解析器来识别每一行的文本，它的作用是将每一行文本转换为数据实体，数据实体这里偷了个懒，字段全部设置成了public。代码片段如下：
/**
* 销售数据解释器
* 销售数据格式为
* userId$countDate(yyyy-MM-dd)$saleCount
*/
public class SaleDataParse implements TextParse<SaleDataEntity> {
@Override
public SaleDataEntity parse(String text) {
if (text == null) {
return null;
}
text = text.trim();
if (text.isEmpty()) {
return null;
}
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
String[] split = text.split("\\$");
SaleDataEntity data = new SaleDataEntity();
data.userId = Long.valueOf(split[0]);
data.countDate = sdf.parse(split[1], new ParsePosition(0));
data.saleCount = Integer.valueOf(split[2]);
return data;
}
}
/**
* 销售数据实体
*/
public class SaleDataEntity {
/**
* 销售用户id
*/
public Long userId;
/**
* 销售日期
*/
public Date countDate;
/**
* 销售总数
*/
public Integer saleCount;
}
2、Map函数
Mapper是一个泛型类，它需要4个泛型参数，从左到右分别是输入键、输入值、输出键和输出值。也就是这样
Mapper<输入键, 输入值, 输出键, 输出值>
其中输入键和输入值的格式是由InputFormatClass决定的，关于输入格式的讨论之后会展开讨论。MapReduce默认会把文件按行拆分，然后偏移量（输入键）->行文本（输入值）的映射传递给Mapper的map方法。输出键和输出值则由用户进行指定。
这里由于是找每一个用户的最大销售数量，Mapper的功能是接收并解析每行数据。所以输出键我设成了销售人员id->销售数量的映射。所以实际的Mapper实现看起来像这样：
/**
* 解析输入的文本数据
*/
public class MaxSaleMapper extends Mapper<LongWritable, Text, LongWritable, IntWritable> {
protected TextParse<SaleDataEntity> saleDataParse = new SaleDataParse();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String s = value.toString();
SaleDataEntity data = saleDataParse.parse(s);
if (data != null) {
//写入输出给Reducer
context.write(new LongWritable(data.userId), new IntWritable(data.saleCount));
}
}
}
其中LongWritable相当于java里的long，Text相当于java里的String，IntWritable相当于java里的int。
这里你可能会想到，既然已经解析成了数据实体，为什么不直接把实体设置成输出值？因为map函数和reduce函数不一定运行在同一个进程里，所以会涉及到序列化和反序列化，这里先不展开。
3、Reduce函数
Reducer也是一个泛型类，它也需要4个参数，从左到右分别是输入键、输入值、输出键和输出值。也就是这样
Reducer<输入键, 输入值, 输出键, 输出值>
与Mapper不同的是，输入键和输入值来源于Mapper的输出，也就是Mapper实现中的context.write()。
输出键和输出值也是由用户指定，默认的输出会写到文件中，关于Reducer的输出以后会讨论。
Reducer的功能是寻找每个用户的最大值，所以Reducer的实现看起来像这样：
/**
* 查找每一个用户的最大销售值
*/
public class MaxSaleReducer extends Reducer<LongWritable, IntWritable, LongWritable, IntWritable> {
@Override
protected void reduce(LongWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
int max = 0;
for (IntWritable value : values) {
if (value.get() > max) {
max = value.get();
}
}
context.write(key, new IntWritable(max));
}
}
你可能会奇怪，为什么reduce方法的第二个参数是一个迭代器。简单来说，Mapper会把映射的值进行归并，然后再传递给Reducer。
4、驱动程序
我们已经完成了map和reduce函数的实现，现在我们需要把它们组装起来。我们需要写一个Main类，它看起来像这样
public class MaxSale {
public static void main(String[] args) throws Exception {
Job job = Job.getInstance();
job.setJarByClass(MaxSale.class);
job.setJobName("MaxSale");
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setMapperClass(MaxSaleMapper.class);
job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(IntWritable.class);
job.setReducerClass(MaxSaleReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
//设置Reduce任务数
job.setNumReduceTasks(1);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
这里解释一下
- 首先我们创建了一个Job
- 然后设置输入目录和输出目录，它们分别是FileInputFormat.addInputPath和FileOutputFormat.setOutputPath
- 使用setMapperClass设置了map函数，setMapOutputKeyClass设置了map函数的输入键类型，setMapOutputValueClass设置了输出键类型
- 使用setReducerClass设置了reduce函数，setOutputKeyClass设置了输出键类型，setOutputValueClass设置了输出值类型
- 然后使用setNumReduceTasks设置reduce任务个数为1，每个reduce任务都会输出一个文件，这里是为了方便查看
- 最后job.waitForCompletion(true)启动并等待任务结束
5、运行结果
使用maven package打包，会生成一个jar，我生成的名字是maxSaleMapReduce-1.0-SNAPSHOT.jar。如果打包的jar有除了Hadoop的其他依赖，需要设置一下HADOOP_CLASSPATH，然后把依赖放到HADOOP_CLASSPATH目录中。
最后输入启动命令，格式为：hadoop jar 生成的jar.jar 输入数据目录 输出数据目录。这里给出我使用的命令示例：
Windows：
set HADOOP_CLASSPATH=C:\xxxxxxxxx\lib\*
hadoop jar maxSaleMapReduce-1.0-SNAPSHOT.jar input output
然后你会看到程序有如下输出，这里截取的部分：
23/01/18 12:10:29 INFO mapred.MapTask: Starting flush of map output
23/01/18 12:10:29 INFO mapred.MapTask: Spilling map output
23/01/18 12:10:29 INFO mapred.MapTask: bufstart = 0; bufend = 17677320; bufvoid = 104857600
23/01/18 12:10:29 INFO mapred.MapTask: kvstart = 26214396(104857584); kvend = 20321960(81287840); length = 5892437/6553600
23/01/18 12:10:30 INFO mapred.MapTask: Finished spill 0
23/01/18 12:10:30 INFO mapred.Task: Task:attempt_local1909247000_0001_m_000000_0 is done. And is in the process of committing
23/01/18 12:10:30 INFO mapred.LocalJobRunner: map
23/01/18 12:10:30 INFO mapred.Task: Task 'attempt_local1909247000_0001_m_000000_0' done.
23/01/18 12:10:30 INFO mapred.Task: Final Counters for attempt_local1909247000_0001_m_000000_0: Counters: 17
File System Counters
FILE: Number of bytes read=33569210
FILE: Number of bytes written=21132276
FILE: Number of read operations=0
FILE: Number of large read operations=0
FILE: Number of write operations=0
Map-Reduce Framework
Map input records=1473110
Map output records=1473110
Map output bytes=17677320
Map output materialized bytes=20623546
Input split bytes=122
Combine input records=0
Spilled Records=1473110
Failed Shuffles=0
Merged Map outputs=0
GC time elapsed (ms)=36
Total committed heap usage (bytes)=268435456
File Input Format Counters
Bytes Read=33558528
23/01/18 12:10:30 INFO mapred.LocalJobRunner: Finishing task: attempt_local1909247000_0001_m_000000_0
23/01/18 12:10:30 INFO mapred.LocalJobRunner: Starting task: attempt_local1909247000_0001_m_000001_0
23/01/18 12:10:30 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 1
23/01/18 12:10:30 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false
等待程序执行结束，output文件夹会有输出part-r-00000，文件里每一行是每一个用户的id和他销售最大值。
0 9994
1 9975
2 9987
3 9985
4 9978
5 9998
三、MapReduce执行流程
简单分析一下这个示例程度的执行流程：
- 首先输入文件被按行切分，输入到各个maper
- maper的输出按输出键进行分类，经过shuffle操作后输入到reducer
- reducer收到maper的输出后，执行寻找最大值操作，然后输出
- 输出会被默认的输出格式格式化后输出到文件part-r-00000中
四、示例代码说明
本文所有的代码放在我的github上，地址是：https://github.com/xunpengliu/hello-hadoop
下面是项目目录说明：
- maxSaleMapReduce模块是Map函数和Reduce的实现，这个模块依赖common模块。所以运行的时候需要把common模块生成的jar添加到HADOOP_CLASSPATH中
- common模块是公共模块，里面有一个SaleDataGenerator的数据生成器，可以生成本次示例代码使用的生成数据
最后需要说明的是，项目代码主要用于学习，代码风格并非代表本人实际风格，不完善之处请轻喷。
五、常见问题
-
java.lang.RuntimeException: java.io.FileNotFoundException: Could not locate Hadoop executable: xxxxxxxxxxxx\bin\winutils.exe -see https://wiki.apache.org/hadoop/WindowsProblems
这个是因为没有下载winutils.exe和hadoop.dll，具体可以参考《安装一个最小化的Hadoop》中windows额外说明
-
运行出现异常java.lang.NullPointerException
at java.lang.ProcessBuilder.start(ProcessBuilder.java:1012)
at org.apache.hadoop.util.Shell.runCommand(Shell.java:482)
at org.apache.hadoop.util.Shell.run(Shell.java:455)
at org.apache.hadoop.util.Shell$ShellCommandExecutor.execute(Shell.java:702) ......
这个和问题1类似，Hadoop在Windows需要winutils.exe和hadoop.dll访问文件，这两个文件通过org.apache.hadoop.util.Shell#getQualifiedBinPath这个方法获取，而这个方法又依赖Hadoop的安装目录。
设置HADOOP_HOME环境变量，或者传入系统参数hadoop.home.dir为Hadoop程序目录，具体参见《安装一个最小化的Hadoop》