做防护用品的网站,公司注册网站模板,视频网站搭建源码,南宁app小程序开发公司转载自 https://www.cnblogs.com/gaopeng527/p/5436820.html 1、MapReduce理论简介 1.1 MapReduce编程模型 MapReduce采用分而治之的思想#xff0c;把对大规模数据集的操作#xff0c;分发给一个主节点管理下的各个分节点共同完成#xff0c;然后通过整合各个节…转载自 https://www.cnblogs.com/gaopeng527/p/5436820.html 1、MapReduce理论简介 1.1 MapReduce编程模型 MapReduce采用分而治之的思想把对大规模数据集的操作分发给一个主节点管理下的各个分节点共同完成然后通过整合各个节点的中间结果得到最终结果。简单地说MapReduce就是任务的分解与结果的汇总。 在Hadoop中用于执行MapReduce任务的机器角色有两个一个是JobTracker另一个是TaskTrackerJobTracker是用于调度工作的TaskTracker是用于执行工作的。一个Hadoop集群中只有一台JobTracker。 在分布式计算中MapReduce框架负责处理了并行编程中分布式存储、工作调度、负载均衡、容错均衡、容错处理以及网络通信等复杂问题把处理过程高度抽象为两个函数map和reducemap负责把任务分解成多个任务reduce负责把分解后多任务处理的结果汇总起来。 需要注意的是用MapReduce来处理的数据集或任务必须具备这样的特点待处理的数据集可以分解成许多小的数据集而且每一个小数据集都可以完全并行地进行处理。 1.2 MapReduce处理过程 在Hadoop中每个MapReduce任务都被初始化为一个Job每个Job又可以分为两种阶段map阶段和reduce阶段。这两个阶段分别用两个函数表示即map函数和reduce函数。map函数接收一个key,value形式的输入然后同样产生一个key,value形式的中间输出Hadoop函数接收一个如key,(list of values)形式的输入然后对这个value集合进行处理每个reduce产生0或1个输出reduce的输出也是key,value形式的。 MapReduce处理大数据集的过程 2、运行WordCount程序 单词计数是最简单也是最能体现MapReduce思想的程序之一可以称为MapReduce版Hello World该程序的完整代码可以在Hadoop安装包的src/examples目录下找到。单词计数主要完成功能是统计一系列文本文件中每个单词出现的次数如下图所示。 2.1 准备工作 现在以hadoop普通用户登录Master.Hadoop服务器。 1创建本地示例文件 首先在/home/hadoop目录下创建文件夹file。 接着创建两个文本文件file1.txt和file2.txt使file1.txt内容为Hello World而file2.txt的内容为Hello Hadoop。 2在HDFS上创建输入文件夹 3上传本地file中文件到集群的input目录下 2.2 运行例子 1在集群上运行WordCount程序 备注以input作为输入目录output目录作为输出目录。 已经编译好的WordCount的Jar在/usr/hadoop下面就是hadoop-examples-1.0.0.jar所以在下面执行命令时记得把路径写全了不然会提示找不到该Jar包。 2MapReduce执行过程显示信息 Hadoop命令会启动一个JVM来运行这个MapReduce程序并自动获得Hadoop的配置同时把类的路径及其依赖关系加入到Hadoop的库中。以上就是Hadoop Job的运行记录从这里可以看到这个Job被赋予了一个ID号job_201202292213_0002而且得知输入文件有两个Total input paths to process : 2同时还可以了解map的输入输出记录record数及字节数以及reduce输入输出记录。比如说在本例中map的task数量是2个reduce的task数量是一个。map的输入record数是2个输出record数是4个等信息。 2.3 查看结果 1查看HDFS上output目录内容 从上图中知道生成了三个文件我们的结果在part-r-00000中。 2查看结果输出文件内容 3、WordCount源码分析 3.1 特别数据类型介绍 Hadoop提供了如下内容的数据类型这些数据类型都实现了WritableComparable接口以便用这些类型定义的数据可以被序列化进行网络传输和文件存储以及进行大小比较。 BooleanWritable标准布尔型数值 ByteWritable单字节数值 DoubleWritable双字节数 FloatWritable浮点数 IntWritable整型数 LongWritable长整型数 Text使用UTF8格式存储的文本 NullWritable当key,value中的key或value为空时使用 3.2 旧的WordCount分析 1源代码程序 package org.apache.hadoop.examples; import java.io.IOException; import java.util.Iterator; import java.util.StringTokenizer; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapred.FileInputFormat; import org.apache.hadoop.mapred.FileOutputFormat; import org.apache.hadoop.mapred.JobClient; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.MapReduceBase; import org.apache.hadoop.mapred.Mapper; import org.apache.hadoop.mapred.OutputCollector; import org.apache.hadoop.mapred.Reducer; import org.apache.hadoop.mapred.Reporter; import org.apache.hadoop.mapred.TextInputFormat; import org.apache.hadoop.mapred.TextOutputFormat; public class WordCount { public static class Map extends MapReduceBase implements MapperLongWritable, Text, Text, IntWritable { private final static IntWritable one new IntWritable(1); private Text word new Text(); public void map(LongWritable key, Text value, OutputCollectorText, IntWritable output, Reporter reporter) throws IOException { String line value.toString(); StringTokenizer tokenizer new StringTokenizer(line); while (tokenizer.hasMoreTokens()) { word.set(tokenizer.nextToken()); output.collect(word, one); } } } public static class Reduce extends MapReduceBase implements ReducerText, IntWritable, Text, IntWritable { public void reduce(Text key, IteratorIntWritable values, OutputCollectorText, IntWritable output, Reporter reporter) throws IOException { int sum 0; while (values.hasNext()) { sum values.next().get(); } output.collect(key, new IntWritable(sum)); } } public static void main(String[] args) throws Exception { JobConf conf new JobConf(WordCount.class); conf.setJobName(wordcount); conf.setOutputKeyClass(Text.class); conf.setOutputValueClass(IntWritable.class); conf.setMapperClass(Map.class); conf.setCombinerClass(Reduce.class); conf.setReducerClass(Reduce.class); conf.setInputFormat(TextInputFormat.class); conf.setOutputFormat(TextOutputFormat.class); FileInputFormat.setInputPaths(conf, new Path(args[0])); FileOutputFormat.setOutputPath(conf, new Path(args[1])); JobClient.runJob(conf); } } 3主方法Main分析 public static void main(String[] args) throws Exception { JobConf conf new JobConf(WordCount.class); conf.setJobName(wordcount); conf.setOutputKeyClass(Text.class); conf.setOutputValueClass(IntWritable.class); conf.setMapperClass(Map.class); conf.setCombinerClass(Reduce.class); conf.setReducerClass(Reduce.class); conf.setInputFormat(TextInputFormat.class); conf.setOutputFormat(TextOutputFormat.class); FileInputFormat.setInputPaths(conf, new Path(args[0])); FileOutputFormat.setOutputPath(conf, new Path(args[1])); JobClient.runJob(conf); } 首先讲解一下Job的初始化过程。main函数调用Jobconf类来对MapReduce Job进行初始化然后调用setJobName()方法命名这个Job。对Job进行合理的命名有助于更快地找到Job以便在JobTracker和Tasktracker的页面中对其进行监视。 JobConf conf new JobConf(WordCount. class ); conf.setJobName(wordcount ); 接着设置Job输出结果key,value的中key和value数据类型因为结果是单词,个数所以key设置为Text类型相当于Java中String类型。Value设置为IntWritable相当于Java中的int类型。 conf.setOutputKeyClass(Text.class ); conf.setOutputValueClass(IntWritable.class ); 然后设置Job处理的Map拆分、Combiner中间结果合并以及Reduce合并的相关处理类。这里用Reduce类来进行Map产生的中间结果合并避免给网络数据传输产生压力。 conf.setMapperClass(Map.class ); conf.setCombinerClass(Reduce.class ); conf.setReducerClass(Reduce.class ); 接着就是调用setInputPath()和setOutputPath()设置输入输出路径。 conf.setInputFormat(TextInputFormat.class ); conf.setOutputFormat(TextOutputFormat.class ); 1InputFormat和InputSplit InputSplit是Hadoop定义的用来传送给每个单独的map的数据InputSplit存储的并非数据本身而是一个分片长度和一个记录数据位置的数组。生成InputSplit的方法可以通过InputFormat()来设置。 当数据传送给map时map会将输入分片传送到InputFormatInputFormat则调用方法getRecordReader()生成RecordReaderRecordReader再通过creatKey()、creatValue()方法创建可供map处理的key,value对。简而言之InputFormat()方法是用来生成可供map处理的key,value对的。 Hadoop预定义了多种方法将不同类型的输入数据转化为map能够处理的key,value对它们都继承自InputFormat分别是 InputFormat | |---BaileyBorweinPlouffe.BbpInputFormat |---ComposableInputFormat |---CompositeInputFormat |---DBInputFormat |---DistSum.Machine.AbstractInputFormat |---FileInputFormat |---CombineFileInputFormat |---KeyValueTextInputFormat |---NLineInputFormat |---SequenceFileInputFormat |---TeraInputFormat |---TextInputFormat 其中TextInputFormat是Hadoop默认的输入方法在TextInputFormat中每个文件或其一部分都会单独地作为map的输入而这个是继承自FileInputFormat的。之后每行数据都会生成一条记录每条记录则表示成key,value形式 key值是每个数据的记录在数据分片中字节偏移量数据类型是LongWritable
value值是每行的内容数据类型是Text。 2OutputFormat 每一种输入格式都有一种输出格式与其对应。默认的输出格式是TextOutputFormat这种输出方式与输入类似会将每条记录以一行的形式存入文本文件。不过它的键和值可以是任意形式的因为程序内容会调用toString()方法将键和值转换为String类型再输出。 3Map类中map方法分析 public static class Map extends MapReduceBase implements MapperLongWritable, Text, Text, IntWritable { private final static IntWritable one new IntWritable(1); private Text word new Text(); public void map(LongWritable key, Text value, OutputCollectorText, IntWritable output, Reporter reporter) throws IOException { String line value.toString(); StringTokenizer tokenizer new StringTokenizer(line); while (tokenizer.hasMoreTokens()) { word.set(tokenizer.nextToken()); output.collect(word, one); } } } Map类继承自MapReduceBase并且它实现了Mapper接口此接口是一个规范类型它有4种形式的参数分别用来指定map的输入key值类型、输入value值类型、输出key值类型和输出value值类型。在本例中因为使用的是TextInputFormat它的输出key值是LongWritable类型输出value值是Text类型所以map的输入类型为LongWritable,Text。在本例中需要输出word,1这样的形式因此输出的key值类型是Text输出的value值类型是IntWritable。 实现此接口类还需要实现map方法map方法会具体负责对输入进行操作在本例中map方法对输入的行以空格为单位进行切分然后使用OutputCollect收集输出的word,1。 4Reduce类中reduce方法分析 public static class Reduce extends MapReduceBase implements ReducerText, IntWritable, Text, IntWritable { public void reduce(Text key, IteratorIntWritable values, OutputCollectorText, IntWritable output, Reporter reporter) throws IOException { int sum 0; while (values.hasNext()) { sum values.next().get(); } output.collect(key, new IntWritable(sum)); } } Reduce类也是继承自MapReduceBase的需要实现Reducer接口。Reduce类以map的输出作为输入因此Reduce的输入类型是TextIntwritable。而Reduce的输出是单词和它的数目因此它的输出类型是Text,IntWritable。Reduce类也要实现reduce方法在此方法中reduce函数将输入的key值作为输出的key值然后将获得多个value值加起来作为输出的值。 3.3 新的WordCount分析 1源代码程序 package org.apache.hadoop.examples; import java.io.IOException; import java.util.StringTokenizer; 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.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class WordCount { public static class TokenizerMapper extends MapperObject, Text, Text, IntWritable{ private final static IntWritable one new IntWritable(1); private Text word new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } } public static class IntSumReducer extends ReducerText,IntWritable,Text,IntWritable { private IntWritable result new IntWritable(); public void reduce(Text key, IterableIntWritable values,Context context) throws IOException, InterruptedException { int sum 0; for (IntWritable val : values) { sum val.get(); } result.set(sum); context.write(key, result); } } public static void main(String[] args) throws Exception { Configuration conf new Configuration(); String[] otherArgs new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length ! 2) { System.err.println(Usage: wordcount in out); System.exit(2); } Job job new Job(conf, word count); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } } 1Map过程 public static class TokenizerMapper extends MapperObject, Text, Text, IntWritable{ private final static IntWritable one new IntWritable(1); private Text word new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } Map过程需要继承org.apache.hadoop.mapreduce包中Mapper类并重写其map方法。通过在map方法中添加两句把key值和value值输出到控制台的代码可以发现map方法中value值存储的是文本文件中的一行以回车符为行结束标记而key值为该行的首字母相对于文本文件的首地址的偏移量。然后StringTokenizer类将每一行拆分成为一个个的单词并将word,1作为map方法的结果输出其余的工作都交有MapReduce框架处理。 2Reduce过程 public static class IntSumReducer extends ReducerText,IntWritable,Text,IntWritable { private IntWritable result new IntWritable(); public void reduce(Text key, IterableIntWritable values,Context context) throws IOException, InterruptedException { int sum 0; for (IntWritable val : values) { sum val.get(); } result.set(sum); context.write(key, result); } } Reduce过程需要继承org.apache.hadoop.mapreduce包中Reducer类并重写其reduce方法。Map过程输出key,values中key为单个单词而values是对应单词的计数值所组成的列表Map的输出就是Reduce的输入所以reduce方法只要遍历values并求和即可得到某个单词的总次数。 3执行MapReduce任务 public static void main(String[] args) throws Exception { Configuration conf new Configuration(); String[] otherArgs new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length ! 2) { System.err.println(Usage: wordcount in out); System.exit(2); } Job job new Job(conf, word count); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } 在MapReduce中由Job对象负责管理和运行一个计算任务并通过Job的一些方法对任务的参数进行相关的设置。此处设置了使用TokenizerMapper完成Map过程中的处理和使用IntSumReducer完成Combine和Reduce过程中的处理。还设置了Map过程和Reduce过程的输出类型key的类型为Textvalue的类型为IntWritable。任务的输出和输入路径则由命令行参数指定并由FileInputFormat和FileOutputFormat分别设定。完成相应任务的参数设定后即可调用job.waitForCompletion()方法执行任务。 4、WordCount处理过程 本节将对WordCount进行更详细的讲解。详细执行步骤如下 1将文件拆分成splits由于测试用的文件较小所以每个文件为一个split并将文件按行分割形成key,value对如图4-1所示。这一步由MapReduce框架自动完成其中偏移量即key值包括了回车所占的字符数Windows和Linux环境会不同。 图4-1 分割过程 2将分割好的key,value对交给用户定义的map方法进行处理生成新的key,value对如图4-2所示。 图4-2 执行map方法 3得到map方法输出的key,value对后Mapper会将它们按照key值进行排序并执行Combine过程将key至相同value值累加得到Mapper的最终输出结果。如图4-3所示。 图4-3 Map端排序及Combine过程 4Reducer先对从Mapper接收的数据进行排序再交由用户自定义的reduce方法进行处理得到新的key,value对并作为WordCount的输出结果如图4-4所示。 图4-4 Reduce端排序及输出结果 5、MapReduce新旧改变 Hadoop最新版本的MapReduce Release 0.20.0的API包括了一个全新的Mapreduce JAVA API有时候也称为上下文对象。 新的API类型上不兼容以前的API所以以前的应用程序需要重写才能使新的API发挥其作用 。 新的API和旧的API之间有下面几个明显的区别。 新的API倾向于使用抽象类而不是接口因为这更容易扩展。例如你可以添加一个方法(用默认的实现)到一个抽象类而不需修改类之前的实现方法。在新的API中Mapper和Reducer是抽象类。 新的API是在org.apache.hadoop.mapreduce包(和子包)中的。之前版本的API则是放在org.apache.hadoop.mapred中的。 新的API广泛使用context object(上下文对象)并允许用户代码与MapReduce系统进行通信。例如MapContext基本上充当着JobConf的OutputCollector和Reporter的角色。 新的API同时支持推和拉式的迭代。在这两个新老API中键/值记录对被推mapper中但除此之外新的API允许把记录从map()方法中拉出这也适用于reducer。拉式的一个有用的例子是分批处理记录而不是一个接一个。 新的API统一了配置。旧的API有一个特殊的JobConf对象用于作业配置这是一个对于Hadoop通常的Configuration对象的扩展。在新的API中这种区别没有了所以作业配置通过Configuration来完成。作业控制的执行由Job类来负责而不是JobClient它在新的API中已经荡然无存。
