• 1. 北京传智播客教育 www.itcast.cnHadoop深入浅出讲师: 吴 超 博客:www.superwu.cn Q Q: 3774 86624
  • 2. 课程安排MapReduce原理*** MapReduce执行过程** 数据类型与格式*** Writable接口与序列化机制*** ---------------------------加深拓展---------------------- MapReduce的执行过程源码分析
  • 3. MapReduce概述◆MapReduce是一种分布式计算模型,由Google提出,主要用于搜索领域,解决海量数据的计算问题. ◆MR由两个阶段组成:Map和Reduce,用户只需要实现map()和reduce()两个函数,即可实现分布式计算,非常简单。 ◆这两个函数的形参是key、value对,表示函数的输入信息。
  • 4. MR执行流程
  • 5. Mapreduce原理
  • 6. ◆执行步骤: 1. map任务处理 1.1 读取输入文件内容,解析成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、value处理,转换成新的key、value输出。 2.3 把reduce的输出保存到文件中。 例子:实现WordCountApp
  • 7. map、reduce键值对格式函数输入键值对输出键值对map()reduce()
  • 8. WordCountApp的驱动代码public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); //加载配置文件 Job job = new Job(conf); //创建一个job,供JobTracker使用 job.setJarByClass(WordCountApp.class); job.setMapperClass(WordCountMapper.class); job.setReducerClass(WordCountReducer.class); FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.1.10:9000/input")); FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.1.10:9000/output")); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.waitForCompletion(true); }
  • 9. JobTracker负责接收用户提交的作业,负责启动、跟踪任务执行。 JobSubmissionProtocol是JobClient与JobTracker通信的接口。 InterTrackerProtocol是TaskTracker与JobTracker通信的接口。
  • 10. TaskTracker负责执行任务。
  • 11. JobClient是用户作业与JobTracker交互的主要接口。 负责提交作业的,负责启动、跟踪任务执行、访问任务状态和日志等。
  • 12. (本页无文本内容)
  • 13. 最小的MapReduce驱动Configuration configuration = new Configuration(); 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); job.setReducerClass(IdentityReducer.class); job.setOutputKeyClass(LongWritable.class); job.setOutputValueClass(Text.class); job.setOutputFormat(TextOutputFormat.class); job.waitForCompletion(true);
  • 14. MapReduce驱动默认的设置InputFormat(输入)TextInputFormatMapperClass(map类)IdentityMapperMapOutputKeyClassLongWritableMapOutputValueClassTextPartitionerClassHashPartitionerReduceClassIdentityReduceOutputKeyClassLongWritableOutputValueClassTextOutputFormatClassTextOutputFormat
  • 15. 序列化概念序列化(Serialization)是指把结构化对象转化为字节流。 反序列化(Deserialization)是序列化的逆过程。即把字节流转回结构化对象。 Java序列化(java.io.Serializable)
  • 16. Hadoop序列化的特点序列化格式特点: 紧凑:高效使用存储空间。 快速:读写数据的额外开销小 可扩展:可透明地读取老格式的数据 互操作:支持多语言的交互 Hadoop的序列化格式:Writable
  • 17. Hadoop序列化的作用序列化在分布式环境的两大作用:进程间通信,永久存储。 Hadoop节点间通信。消息 序列化为 二进制流节点1二进制流 反序列化为 消息节点2二进制流消息
  • 18. Writable接口Writable接口, 是根据 DataInput 和 DataOutput 实现的简单、有效的序列化对象. MR的任意Key和Value必须实现Writable接口.MR的任意key必须实现WritableComparable接口
  • 19. 常用的Writable实现类Text一般认为它等价于java.lang.String的Writable。针对UTF-8序列。 例: Text test = new Text("test"); IntWritable one = new IntWritable(1);
  • 20. 自定义Writable类实现WritableComparable. Java值对象的比较:一般需要重写toString(),hashCode(), equals()方法 Writable write 是把每个对象序列化到输出流 readFields是把输入流字节反序列化
  • 21. 基于文件的存储结构SequenceFile 无序存储 MapFile 会对key建立索引文件,value按key顺序存储 基于MapFile的结构有: ArrayFile 像我们使用的数组一样,key值为序列化的数字 SetFile 他只有key,value为不可变的数据 BloomMapFile 在 MapFile 的基础上增加了一个 /bloom 文件,包含的是二进制的过滤表,在每一次写操作完成时,会更新这个过滤表。
  • 22. 自定义WritableKpi电信例子 把上面例子里的Mapper的value改写为自定义Writable类型。修改原MapReduce程序,并成功执行。结果跟原来一致。
  • 23. MapReduce输入的处理类 FileInputFormat:  FileInputFormat是所有以文件作为数据源的InputFormat实现的基类,FileInputFormat保存作为job输入的所有文件,并实现了对输入文件计算splits的方法。至于获得记录的方法是有不同的子类——TextInputFormat进行实现的。 
  • 24. InputFormatInputFormat 负责处理MR的输入部分. 有三个作用: 验证作业的输入是否规范. 把输入文件切分成InputSplit. 提供RecordReader 的实现类,把InputSplit读到Mapper中进行处理.
  • 25. InputSplit◆ 在执行mapreduce之前,原始数据被分割成若干split,每个split作为一个map任务的输入,在map执行过程中split会被分解成一个个记录(key-value对),map会依次处理每一个记录。 ◆ FileInputFormat只划分比HDFS block大的文件,所以FileInputFormat划分的结果是这个文件或者是这个文件中的一部分. ◆ 如果一个文件的大小比block小,将不会被划分,这也是Hadoop处理大文件的效率要比处理很多小文件的效率高的原因。 ◆ 当Hadoop处理很多小文件(文件大小小于hdfs block大小)的时候,由于FileInputFormat不会对小文件进行划分,所以每一个小文件都会被当做一个split并分配一个map任务,导致效率底下。 例如:一个1G的文件,会被划分成16个64MB的split,并分配16个map任务处理,而10000个100kb的文件会被10000个map任务处理。   
  • 26. TextInputFormat◆ TextInputformat是默认的处理类,处理普通文本文件。 ◆ 文件中每一行作为一个记录,他将每一行在文件中的起始偏移量作为key,每一行的内容作为value。 ◆ 默认以\n或回车键作为一行记录。 ◆ TextInputFormat继承了FileInputFormat。
  • 27. InputFormat类的层次结构
  • 28. 其他输入类◆ CombineFileInputFormat 相对于大量的小文件来说,hadoop更合适处理少量的大文件。 CombineFileInputFormat可以缓解这个问题,它是针对小文件而设计的。 ◆ KeyValueTextInputFormat 当输入数据的每一行是两列,并用tab分离的形式的时候,KeyValueTextInputformat处理这种格式的文件非常适合。 ◆ NLineInputformat  NLineInputformat可以控制在每个split中数据的行数。 ◆ SequenceFileInputformat  当输入文件格式是sequencefile的时候,要使用SequenceFileInputformat作为输入。  
  • 29. 自定义输入格式1)继承FileInputFormat基类。 2)重写里面的getSplits(JobContext context)方法。 3)重写createRecordReader(InputSplit split, TaskAttemptContext context)方法。 (讲解源代码)
  • 30. Hadoop的输出◆ TextOutputformat 默认的输出格式,key和value中间值用tab隔开的。 ◆ SequenceFileOutputformat 将key和value以sequencefile格式输出。 ◆ SequenceFileAsOutputFormat 将key和value以原始二进制的格式输出。 ◆ MapFileOutputFormat 将key和value写入MapFile中。由于MapFile中的key是有序的,所以写入的时候必须保证记录是按key值顺序写入的。 ◆ MultipleOutputFormat 默认情况下一个reducer会产生一个输出,但是有些时候我们想一个reducer产生多个输出,MultipleOutputFormat和MultipleOutputs可以实现这个功能。
  • 31. 思考题MapReduce框架的结构是什么 Map在整个MR框架中作用是什么 Reduce在整个MR框架中作用是什么