JunLeon——go big or go home

目录 

MapReduce工作原理及基础编程（代码见文章后半部分）

一、MapReduce概述

1、什么是MapReduce？

2、WordCount案例解析MapReduce计算过程

（1）运行hadoop自带的样例程序

（2）MapReduce工作过程

3、Shuffle过程详解

二、MapReduce编程基础

1、Hadoop数据类型

2、数据输入格式InputFormat

3、输入数据分块InputSplit和数据记录读入RecordReader

4、数据输出格式OutputFormat

5、数据记录输出类RecordWriter

 6、Mapper类

7、Reduce类

三、MapReduce项目案例

1、经典案例——WordCount

2、计算考试平均成绩

3、网站日志分析

 前言：

        Google于2003年在SOSP上发表了《The Google File System》，于2004年在OSDI上发表了《MapReduce: Simplified Data Processing on Large Clusters》，于2006年在OSDI上发表了《Bigtable: A Distributed Storage System for Structured Data》。这三篇论文为大数据及云计算的发展奠定了基础。

        MapReduce是一个分布式、并行处理的计算框架。

        MapReduce 把任务分为 Map 阶段和 Reduce 阶段。开发人员使用存储在HDFS 中数据（可实现快速存储），编写 Hadoop 的 MapReduce 任务。由于 MapReduce工作原理的特性， Hadoop 能以并行的方式访问数据，从而实现快速访问数据。

表1 map函数和rudece函数

List

 MapReduce体系结构主要由四个部分组成，分别是：Client、JobTracker、TaskTracker以及Task

　　1）Client

　　用户编写的MapReduce程序通过Client提交到JobTracker端 用户可通过Client提供的一些接口查看作业运行状态。

　　2）JobTracker

　　JobTracker负责资源监控和作业调度 JobTracker 监控所有TaskTracker与Job的健康状况，一旦发现失败，就将相应的任务转移到其他节点 JobTracker 会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器（TaskScheduler），而调度器会在资源出现空闲时，选择合适的任务去使用这些资源。

　　3）TaskTracker

　　TaskTracker 会周期性地通过“心跳”将本节点上资源的使用情况和任务的运行进度汇报给JobTracker，同时接收JobTracker 发送过来的命令并执行相应的操作（如启动新任务、杀死任务等） TaskTracker 使用“slot”等量划分本节点上的资源量（CPU、内存等）。一个Task 获取到一个slot 后才有机会运行，而Hadoop调度器的作用就是将各个TaskTracker上的空闲slot分配给Task使用。slot 分为Map slot 和Reduce slot 两种，分别供MapTask 和Reduce Task 使用。

　　4）Task

　　Task 分为Map Task 和Reduce Task 两种，均由TaskTracker 启动。

MapReduce各个执行阶段：

 MapReduce应用程序执行过程：

可以参考大佬黎先生的博客：MapReduce基本原理及应用 - 黎先生 - 博客园

WordCount案例是一个经典案例，是Hadoop自带的样例程序。

作用：统计单词数量（出现的次数）

应用：求和、求平均值、求最值，

jar包存储在$HADOOP_HOME/share/hadoop/mapreduce/：

$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.3.jar

例如：

 步骤：

1.在本地创建一个文件

输入以下内容：

2.上传到HDFS指定目录

在HDFS中创建指定文件：

 上传文件：

 3.使用hadoop jar命令运行jar程序，统计单词数量

 4.输出结果

执行部分过程：

 查看生成的文件：

 查看计算结果：

        工作流程是Input从HDFS里面并行读取文本中的内容，经过MapReduce模型，最终把分析出来的结果用Output封装，持久化到HDFS中。

1.Mapper工作过程：

 附上Mapper阶段代码：

KEYIN--LongWritable:输入key类型，记录数据分片的偏移位置

VALUEIN—Text:输入的value类型，对应分片中的文本数据

KEYOUT--Text:输出的key类型，对应map方法中计算结果的key值

VALUEOUT—IntWritable:输出的value类型，对应map方法中计算结果的value值

        Mapper类从分片后传出的上下文中接收数据，数据以类型的键值对接收过来，通过重写map方法默认一行一行的读取数据并且以形式进行遍历赋值。

2.Reducer工作过程：

 附上Reducer阶段代码:

 Reducer任务继承Reducer类，主要接收的数据来自Map任务的输出，中间经过Shuffle分区、排序、分组，最终以形式输出给用户。

Job提交代码：

       JobClients是用户提交的作业与ResourceManager交互的主要接口，JobClients提供提交作业、追踪进程、访问子任务的日志记录、获取的MapReduce集群状态信息等功能。

        Hadoop运行机制中，将map输出进行分区、分组、排序、和合并等处理后作为输入传给Reducer的过程，称为shuffle过程。

 shuffle阶段又可以分为Map端的shuffle和Reduce端的shuffle。

　　一、Map端的shuffle

　　写磁盘：Map端会处理输入数据并产生中间结果，这个中间结果会写到本地磁盘，而不是HDFS。每个Map的输出会先写到内存缓冲区中，当写入的数据达到设定的阈值时，系统将会启动一个线程将缓冲区的数据写到磁盘，这个过程叫做spill。

　　分区、分组、排序：在spill写入之前，会先进行二次排序，首先根据数据所属的partition进行排序，然后每个分区（partition）中的数据再按key来排序。partition的目是将记录划分到不同的Reducer上去，以期望能够达到负载均衡，以后的Reducer就会根据partition来读取自己对应的数据。接着运行combiner(如果设置了的话)，combiner的本质也是一个Reducer，其目的是对将要写入到磁盘上的文件先进行一次处理，这样，写入到磁盘的数据量就会减少。最后将数据写到本地磁盘产生spill文件(spill文件保存在{mapred.local.dir}指定的目录中，Map任务结束后就会被删除)。

        文件合并：最后，每个Map任务可能产生多个溢写文件（spill file），在每个Map任务完成前，会通过多路归并算法将这些spill文件归并成一个已经分区和排序的输出文件。至此，Map的shuffle过程就结束了。

        压缩：在shuffle过程中如果压缩被启用，在map传出数据传入Reduce之前可执行压缩，默认情况下压缩是关闭的，可以将mapred.compress.map.output设置为true可实现压缩。

　　二、Reduce端的shuffle

　　Reduce端的shuffle主要包括三个阶段，copy、sort(merge)和reduce。

　　首先要将Map端产生的输出文件拷贝到Reduce端，但每个Reducer如何知道自己应该处理哪些数据呢？因为Map端进行partition的时候，实际上就相当于指定了每个Reducer要处理的数据(partition就对应了Reducer)，所以Reducer在拷贝数据的时候只需拷贝与自己对应的partition中的数据即可。每个Reducer会处理一个或者多个partition，但需要先将自己对应的partition中的数据从每个Map的输出结果中拷贝过来。

        接下来就是排序（sort）阶段，也成为合并（merge）阶段，因为这个阶段的主要工作是执行了归并排序。从Map端拷贝到Reduce端的数据都是有序的，所以很适合归并排序。最终在Reduce端生成一个较大的文件作为Reduce的输入。MapReduce编程接口

        Hadoop数据包括：BooleanWritable、ByteWritable、DoubleWritable、FloatWritale、IntWritable、LongWritable、Text、NullWritable等，它们实现了WritableComparable接口。其中Text表示使用UTF8格式存储的文本、NullWritable类型是当（key,value）中的key或value为空时使用。

表2 Hadoop Writable与Java数据类型参照表

IntWritable

VIntWritable

4

1-5

LongWritable

        除了上述Hadoop类型外，用户还可以自定义新的数据类型。用户自定义数据类型需要实现Writable接口，但如果需要作为主键key使用或需要比较大小时，则需要实现WritableComparable接口。

抽象类InputFormat有三个直接子类：

        FileInputFormat、DBInputFormat、DelegatingInputFormat

其中，文件输入格式类FileInputFormat类有几个子类：

        TextInputFormat、KeyValueInputFormat、SequenceFileInputFormat、NlineInputFormat、CombineFileInputFormat

序列化文件输入类SequenceFileInputFormat有几个子类：

        SequenceFileAsBinaryInputFormat、SequenceFileAsTextInputFormat、SequenceFileInputFilter

数据库输入格式类DBInputFormat的直接子类是：DataDriverDBInputFormat，而这个子类又派生子类：OracleDataDriverDBInputFormat

表3  常用数据输入格式类

        编程时由用户选择的数据输入格式InputFormat类型来自动决定数据分块InputSplit和数据记录RecordReader类型。一个InputSplit将单独作为一个Mapper的输入，即作业的Mapper数量是由InputSplit个数决定的。

表4 数据输出格式类对应的Reader类型

抽象类OutputFormat有四个直接子类：

        FileOutputFormat、DBOutputFormat、NullOutputFormat、FilterOutputFormat

FileOutputFormat有两个直接子类：

        TextOutputFormat、SequenceFileOutputFormat

SequenceFileOutputFormat有直接子类：SequenceFileAsBinaryOutputFormat

FilterOutputFormat有直接子类：LazyOutputFormat

数据记录输出类RecordWriter是一个抽象类。

表5 数据输出格式类对应的数据记录Writer类型

        Mapper类是一个抽象类，位于hadoop-mapreduce-client-core-2.x.x.jar中，其完整类名是：org.apache.hadoop.mapreduce.Mapper，需派生子类使用，在子类中重写map方法：map(KEYIN key,VALUEIN value,Mapper.Context context)对出入的数据分块每个键值对调用一次。

        Reduce类是一个抽象类，位于hadoop-mapreduce-client-core-2.x.x.jar中，其完整类名是：org.apache.hadoop.mapreduce.Reduce，需派生子类使用，在子类中重写reduce方法：reduce(KEYIN key,Inerable value,Reducer.Context context)对出入的数据分块每个键值对调用一次。

代码演示：

打包上传虚拟机：

步骤：

右键单击项目名 --> 选择 Export --> Java --> JAR file --> Browse...选择存放路径 --> 文件名

命名为wordcount.jar，将打包好的jar包上传到虚拟机中

运行代码：

在本地创建一个文件input.txt

 添加内容：

上传到DHFS中：

使用jar命令执行项目：

如下图：

查看结果： 

代码演示：

Mapper类

Reducer类

Driver类

命名为average.jar ， 将打包好的average.jar上传到虚拟机中

运行代码：

首先准备三个文件 Chinese.txt、Math.txt、English.txt，添加如下内容：

 将文件上传到HDFS的data目录下： 

执行代码：

 查看结果，如下图：

代码演示：

打包上传虚拟机：

运行代码：