Hadoop深入浅出之MapReduce介绍

1、技术创新，变革未来Hadoop深入浅出之MapReduce介绍课程安排MapReduce原理*MapReduce执行过程*数据类型与格式*Writable接口与序列化机制*-加深拓展-MapReduce的执行过程源码分析MapReduce概述MapReduce是一种分布式计算模型，由Google提出，主要用于搜索领域，解决海量数据的计算问题.MR由两个阶段组成：Map和Reduce，用户只需要实现map()和reduce()两个函数，即可实现分布式计算，非常简单。这两个函数的形参是key、value对，表示函数的输入信息。MR执行流程Mapreduce原理执行步骤： 1. map任务处理1.1

2、 读取输入文件内容，解析成key、value对。对输入文件的每一行，解析成key、value对。每一个键值对调用一次map函数。1.2 写自己的逻辑，对输入的key、value处理，转换成新的key、value输出。1.3 对输出的key、value进行分区。1.4 对不同分区的数据，按照key进行排序、分组。相同key的value放到一个集合中。1.5 (可选)分组后的数据进行归约。2.reduce任务处理2.1 对多个map任务的输出，按照不同的分区，通过网络copy到不同的reduce节点。2.2 对多个map任务的输出进行合并、排序。写reduce函数自己的逻辑，对输入的key、val

3、ue处理，转换成新的key、value输出。2.3 把reduce的输出保存到文件中。例子：实现WordCountApp map、reduce键值对格式函数输入键值对输出键值对map()reduce() WordCountApp的驱动代码public static void main(String args) throws Exception Configuration conf = new Configuration(); /加载配置文件 Job job = new Job(conf); /创建一个job，供JobTracker使用 job.setJarByClass(WordCountAp

4、p.class); job.setMapperClass(WordCountMapper.class); job.setReducerClass(WordCountReducer.class); FileInputFormat.setInputPaths(job, new Path(hdfs:/0:9000/input); FileOutputFormat.setOutputPath(job, new Path(hdfs:/0:9000/output); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable

5、.class); job.waitForCompletion(true);JobTracker负责接收用户提交的作业，负责启动、跟踪任务执行。JobSubmissionProtocol是JobClient与JobTracker通信的接口。InterTrackerProtocol是TaskTracker与JobTracker通信的接口。TaskTracker负责执行任务。JobClient是用户作业与JobTracker交互的主要接口。负责提交作业的，负责启动、跟踪任务执行、访问任务状态和日志等。最小的MapReduce驱动Configuration configuration = new Co

6、nfiguration();Job job = new Job(configuration, HelloWorld);job.setInputFormat(TextInputFormat.class);job.setMapperClass(IdentityMapper.class);job.setMapOutputKeyClass(LongWritable.class);job.setMapOutputValueClass(Text.class);job.setPartitionerClass(HashPartitioner.class);job.setNumReduceTasks(1);jo

7、b.setReducerClass(IdentityReducer.class);job.setOutputKeyClass(LongWritable.class);job.setOutputValueClass(Text.class);job.setOutputFormat(TextOutputFormat.class);job.waitForCompletion(true);MapReduce驱动默认的设置InputFormat(输入)TextInputFormatMapperClass(map类)IdentityMapperMapOutputKeyClassLongWritableMap

8、OutputValueClassTextPartitionerClassHashPartitionerReduceClassIdentityReduceOutputKeyClassLongWritableOutputValueClassTextOutputFormatClassTextOutputFormat序列化概念序列化（Serialization）是指把结构化对象转化为字节流。反序列化（Deserialization）是序列化的逆过程。即把字节流转回结构化对象。Java序列化（java.io.Serializable）Hadoop序列化的特点序列化格式特点：紧凑：高效使用存储空间。快速：

9、读写数据的额外开销小可扩展：可透明地读取老格式的数据互操作：支持多语言的交互Hadoop的序列化格式：WritableHadoop序列化的作用序列化在分布式环境的两大作用：进程间通信，永久存储。Hadoop节点间通信。消息序列化为二进制流节点1二进制流反序列化为消息节点2二进制流消息Writable接口Writable接口, 是根据 DataInput 和 DataOutput 实现的简单、有效的序列化对象.MR的任意Key和Value必须实现Writable接口.MR的任意key必须实现WritableComparable接口常用的Writable实现类Text一般认为它等价于java.la

10、ng.String的Writable。针对UTF-8序列。例:Text test = new Text(test);IntWritable one = new IntWritable(1);自定义Writable类实现WritableComparable.Java值对象的比较：一般需要重写toString(),hashCode(),equals()方法Writablewrite 是把每个对象序列化到输出流readFields是把输入流字节反序列化基于文件的存储结构SequenceFile 无序存储MapFile 会对key建立索引文件，value按key顺序存储基于MapFile的结构有:Ar

11、rayFile 像我们使用的数组一样，key值为序列化的数字SetFile 他只有key，value为不可变的数据BloomMapFile 在 MapFile 的基础上增加了一个 /bloom 文件，包含的是二进制的过滤表，在每一次写操作完成时，会更新这个过滤表。自定义WritableKpi电信例子把上面例子里的Mapper的value改写为自定义Writable类型。修改原MapReduce程序，并成功执行。结果跟原来一致。MapReduce输入的处理类FileInputFormat: FileInputFormat是所有以文件作为数据源的InputFormat实现的基类，FileInput

12、Format保存作为job输入的所有文件，并实现了对输入文件计算splits的方法。至于获得记录的方法是有不同的子类TextInputFormat进行实现的。InputFormatInputFormat 负责处理MR的输入部分.有三个作用:验证作业的输入是否规范.把输入文件切分成InputSplit.提供RecordReader 的实现类，把InputSplit读到Mapper中进行处理.InputSplit 在执行mapreduce之前，原始数据被分割成若干split，每个split作为一个map任务的输入，在map执行过程中split会被分解成一个个记录（key-value对），map会依

13、次处理每一个记录。 FileInputFormat只划分比HDFS block大的文件，所以FileInputFormat划分的结果是这个文件或者是这个文件中的一部分. 如果一个文件的大小比block小，将不会被划分，这也是Hadoop处理大文件的效率要比处理很多小文件的效率高的原因。 当Hadoop处理很多小文件（文件大小小于hdfs block大小）的时候，由于FileInputFormat不会对小文件进行划分，所以每一个小文件都会被当做一个split并分配一个map任务，导致效率底下。 例如：一个1G的文件，会被划分成16个64MB的split，并分配16个map任务处理，而10000个

14、100kb的文件会被10000个map任务处理。 TextInputFormat TextInputformat是默认的处理类，处理普通文本文件。 文件中每一行作为一个记录，他将每一行在文件中的起始偏移量作为key，每一行的内容作为value。 默认以n或回车键作为一行记录。 TextInputFormat继承了FileInputFormat。InputFormat类的层次结构其他输入类 CombineFileInputFormat 相对于大量的小文件来说，hadoop更合适处理少量的大文件。 CombineFileInputFormat可以缓解这个问题，它是针对小文件而设计的。 KeyVal

15、ueTextInputFormat 当输入数据的每一行是两列，并用tab分离的形式的时候，KeyValueTextInputformat处理这种格式的文件非常适合。 NLineInputformat NLineInputformat可以控制在每个split中数据的行数。 SequenceFileInputformat 当输入文件格式是sequencefile的时候，要使用SequenceFileInputformat作为输入。自定义输入格式1）继承FileInputFormat基类。2）重写里面的getSplits(JobContext context)方法。3）重写createRecordR

16、eader(InputSplit split, TaskAttemptContext context)方法。(讲解源代码)Hadoop的输出 TextOutputformat 默认的输出格式，key和value中间值用tab隔开的。 SequenceFileOutputformat 将key和value以sequencefile格式输出。 SequenceFileAsOutputFormat 将key和value以原始二进制的格式输出。 MapFileOutputFormat 将key和value写入MapFile中。由于MapFile中的key是有序的，所以写入的时候必须保证记录是按key值

17、顺序写入的。 MultipleOutputFormat 默认情况下一个reducer会产生一个输出，但是有些时候我们想一个reducer产生多个输出，MultipleOutputFormat和MultipleOutputs可以实现这个功能。思考题MapReduce框架的结构是什么Map在整个MR框架中作用是什么Reduce在整个MR框架中作用是什么课程安排标准和自定义计数器*Combiner和Partitioner编程*自定义排序和分组编程*常见的MapReduce算法*-加深拓展-常见大数据处理方法*MapReduce的老api写法import org.apache.hadoop.fs.Pa

18、th;import org.apache.hadoop.mapred.FileInputFormat;import org.apache.hadoop.mapred.FileOutputFormat;import org.apache.hadoop.mapred.JobClient;import org.apache.hadoop.mapred.JobConf;public class HelloWorldApp public static void main(String args) throws Exception final JobConf job = new JobConf(Hello

19、WorldApp.class);FileInputFormat.setInputPaths(job, new Path(/input);FileOutputFormat.setOutputPath(job, new Path(/output);JobClient.runJob(job);MapReduce获取命令行参数public class CacheFileApp extends Configured implements Toolpublic static void main(String args) throws Exception ToolRunner.run(new Configu

20、ration(), new CacheFileApp(), args); Overridepublic int run(String args) throws Exception /. Path in = new Path(args0); Path out = new Path(args1); /.了解计数器hadoop计数器:可以让开发人员以全局的视角来审查程序的运行情况以及各项指标，及时做出错误诊断并进行相应处理。内置计数器（MapReduce相关、文件系统相关和作业调度相关） .也可以通过http:/master:50030/jobdetails.jsp查看Hello you, hell

21、o me的计数器信息 Counters: 19 File Output Format Counters Bytes Written=19/reduce输出到hdfs的字节数 FileSystemCounters FILE_BYTES_READ=481 HDFS_BYTES_READ=38 FILE_BYTES_WRITTEN=81316 HDFS_BYTES_WRITTEN=19 File Input Format Counters Bytes Read=19/map从hdfs读取的字节数 Map-Reduce Framework Map output materialized bytes=4

22、9 Map input records=2 /map读入的记录行数 Reduce shuffle bytes=0 Spilled Records=8 Map output bytes=35 Total committed heap usage (bytes)=266469376 SPLIT_RAW_BYTES=105 Combine input records=0 Reduce input records=4/reduce从map端接收的记录行数 Reduce input groups=3/reduce函数接收的key数量，即归并后的k2数量 Combine output records=0

23、Reduce output records=3/reduce输出的记录行数 Map output records=4/map输出的记录行数自定义计数器与实现Context类调用方法getCounter()org.apache.hadoop.mapreduce.Counter计数器声明1.通过枚举声明context.getCounter(Enum enum) 2.动态声明context.getCounter(String groupName,String counterName) 计数器操作counter.setValue(long value);/设置初始值counter.increment(

24、long incr);/增加计数Combiners编程 每一个map可能会产生大量的输出，combiner的作用就是在map端对输出先做一次合并，以减少传输到reducer的数据量。 combiner最基本是实现本地key的归并，combiner具有类似本地的reduce功能。 如果不用combiner，那么，所有的结果都是reduce完成，效率会相对低下。使用combiner，先完成的map会在本地聚合，提升速度。注意：Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。所以从我的想法来看，Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。Partitioner是partitioner的基类，如果需要定制partitioner也需要继承该类。2. HashPartitioner是mapreduce的默认partitioner。计算方法是which reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks，得到当前的目的reducer。3. (例子以jar形式运行) Partitioner编程在map和reduce阶段进行排序时，比较的是k2。v2是不参与