MapReduce |【大学生博客大赛】

MapReduce |【大学生博客大赛】

1. 什么是MapReduceMapReduce是Google提出的一种分布式计算模型，用于大规模数据集（大于1TB）的并行运算，解决海量数据的计算问题。主要阶段是Map（映射）和Reduce（归约）。（1）Map任务：

读取输入文件内容，解析成key、value对。对输入文件的每一行，解析成key、value对。每一个键值对调用一次map函数 写自己的逻辑，对输入的key、value处理，转换成新的key、value输出 对输出的key、value进行分区 对相同分区的数据，按照key进行排序(默认按照字典顺序进行排序)，分组。相同key的value放到一个集合中。 (可选)分组后的数据进行归约。 注意：在MapReduce中，Mapper可以单独存在，但是Reducer不能单独存在

(2)Reducer任务：

对多个map任务的输出，按照不同的分区，通过网络copy到不同的reduce节点。这个过程并不是map将数据发送给reduce，而是reduce主动去获取数据。--- Reducer的个数 >= 分区的数量 对多个map任务的输出进行合并、排序。写reduce函数自己的逻辑，对输入的key、value处理，转换成新的key、value输出。 把reduce的输出保存到文件中。

3. MapReduce执行流程（1）做job环境信息的收集，比如各个组件类，输入输出的kv类型等，检测是否合法

4.MapReduce代码的简单实现Mapper代码：

import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class WordCountMapper extends Mapper { @Override protected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException { // TODO WordCount的map函数 // key：偏移量，value：每⼀⾏内容 String line = value.toString(); // 将每⾏的数据按照空格分割 String[] words = line.split(" "); // 将数组中的每个单词，整理成的格式 -> 代表该单词出现⼀次 for (String word : words) { context.write(new Text(word), new IntWritable(1)); } } }

Reducer代码:

import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class WordCountReducer extends Reducer { @Override protected void reduce(Text key, Iterable values, Reducer.Context context) throws IOException, InterruptedException { int count = 0;// 定义⽤于存放结果的变量 for (IntWritable value : values) { count += value.get();// 将value列表中的元素取出，进⾏累加 } context.write(key, new IntWritable(count)); } }

Master代码：

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; public class WordCountMaster { public static void main(String[] args) throws Exception { // TODO wordcount的启动类 Configuration conf = new Configuration(); conf.set("mapreduce.framework.name", "local");// local：本地模式 Job job = Job.getInstance();// 实例化对象：在new之前传⼊参数或做其他操作 job.setJarByClass(WordCountMaster.class);// class相当与Java中的类对象（类的描述 信息：路径、内部结构） // 指定Mapper阶段要执⾏的map函数所在的类 job.setMapperClass(WordCountMapper.class); // Mapper阶段输出的数据结构的类对象 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 指定Reducer阶段要执⾏的reduce函数所在的类 job.setReducerClass(WordCountReducer.class); // Reducer阶段输出的数据结构的类对象（最终结果） job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 本地路径 -> file:///xxx/xxx // HDFS路径 -> hdfs://ip:port/xxx -> /xxx FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 提交运算流程，true:显示详细信息，返回值代表最终的运⾏状态 boolean result = job.waitForCompletion(true); System.out.println(result); } }

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。