03初识MapReduce
初识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);
data.countDate = sdf.parse(split, new ParsePosition(0));
data.saleCount = Integer.valueOf(split);
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<输入键, 输入值, 输出键, 输出值> { protected TextParse 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<输入键, 输入值, 输出键, 输出值> { @Override protected void reduce(LongWritable key, Iterable 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)); FileOutputFormat.setOutputPath(job, new Path(args)); job.setMapper<输入键, 输入值, 输出键, 输出值>Class(MaxSaleMapper<输入键, 输入值, 输出键, 输出值>.class); job.setMapOutputKeyClass(LongWritable.class); job.setMapOutputValueClass(IntWritable.class); job.setReducer<输入键, 输入值, 输出键, 输出值>Class(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
[*]使用setMapper<输入键, 输入值, 输出键, 输出值>Class设置了map函数，setMapOutputKeyClass设置了map函数的输入键类型，setMapOutputValueClass设置了输出键类型
[*]使用setReducer<输入键, 输入值, 输出键, 输出值>Class设置了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执行流程
https://img2023.cnblogs.com/blog/3082749/202301/3082749-20230125202209572-1180966543.png
简单分析一下这个示例程度的执行流程：
[*]首先输入文件被按行切分，输入到各个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》
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
页:
[1]