分布式计算框架Map/Reduce

分布式计算框架MapReduce

1、产生背景

• Web2.0时代，数据爆炸式、指数级增长，大数据分布式计算需求频繁
• 通过单机内存扩展来增强计算能力，已经无法承载大规模数据量的计算
• 分布式计算开发和维护的复杂与多变，对程序员要求太高

Google公司为了解决其搜索引擎中大规模网页数据的并行化处理，研究提出的一种面向大规模数据处理的并行计算模型和方法，称为MapReduce。

2003年和2004年，Google公司在国际会议上分别发表了两篇关于Google分布式文件系统GFS和MapReduce的论文，公布了Google的GFS和MapReduce的基本原理和主要设计思想。

2004年，Cutting和同为程序员出身的Mike Cafarella决定开发一款可以代替当时的主流搜索产品的开源搜索引擎，这个项目被命名为Nutch。2005年初，Nutch的开发人员在Nutch上实现了一个MapReduce系统，到年中，Nutch的所有主要算法均完成移植，用MapReduce和NDFS来运行。在2006年2月，开发人员将NDFS和MapReduce移出Nutch形成Lucene的一个子项目，称为Hadoop。Hadoop中MapReduce的实现正是基于Google的论文的MapReduce的开源实现。

2、MapReduce是什么

MapReduce是一种编程模型，是面向大数据并行处理的计算模型、框架和平台。

• MapReduce是一个基于集群的高性能并行计算平台。可以使用普通服务器构成一个包含数十、数百、甚至数千个节点的分布式和并行计算集群。
• MapReduce是一个并行计算与运行的软件框架。它提供了一个庞大但设计精良的并行计算软件框架，能自动划分计算数据和计算任务，自动完成计算任务的并行化处理，实现在集群节点上自动分配和执行任务并收集计算结果，将数据分布存储、数据通信、容错处理等并行计算涉及到的很多系统底层的复杂实现细节交由系统负责处理，大大减少了软件开发人员的负担。
• MapReduce是一个并行程序设计模型与方法。它提供了一种简便的并行程序设计方法，用Map和Reduce两个函数编程实现基本的并行计算任务，提供了抽象的操作和并行编程接口，以简单方便地完成大规模数据的编程和计算处理。

3、基本特点

• 分布可靠，对数据集的操作分发给集群中的多个节点实现可靠性，每个节点周期性返回它完成的任务和最新的状态
• 封装了实现细节，基于框架API编程，面向业务展开分布式编码
• 提供跨语言编程的能力

4、企业应用

• 各大运营商
• 中大型互联网公司，如BAT、京东、乐视、美团等
• 金融银行保险类公司
• 各大云平台的分布式计算框架
• 其他本地系统无法承载计算能力的应用

MapReduce运行流程

1、MapReduce的主要功能

1.1数据划分和计算任务调度

系统自动将一个作业(Job)待处理的大数据划分为很多个数据块，每个数据块对应于一个计算任务(Task)，并自动调度计算节点来处理相应的数据块。作业和任务调度功能主要负责分配和调度计算节点(Map节点或Reduce节点)，同时负责监控这些节点的执行状态，并负责Map节点执行的同步控制。

1.2数据/代码互相定位

为了减少数据通信，一个基本的原则是本地化数据处理，即一个计算节点尽可能处理其本地磁盘上所分布存储的数据，这实现了代码向数据的迁移；当无法进行这种本地化数据处理时，再寻找其他可用节点并将数据从网络上传送给该节点(数据向代码迁移)，但尽可能从数据所在的本地机架上寻找可用节点以减少通信延迟。

1.3系统优化

为了减少数据通信开销，中间结果数据进入Reduce节点前会进行一定的合并处理；一个Reduce节点所处理的数据可能会来自多个Map节点，为了避免Reduce计算阶段发生数据处理不平衡，Map节点输出的中间结果需使用一定的策略进行适当的划分处理，保证相关性数据发送到同一个Reduce节点；此外，系统还进行一些性能优化处理，如对最慢的计算任务采用多备份执行、选最快完成者作为结果。

1.4出错检测和恢复

以低端的商用服务器构成的大规模MapReduce计算集群中，节点硬件(主机、磁盘、内存等)出错和软件出错是常态，因此MapReduce需要能检测并隔离出错节点，并调度分配新的节点接管出错节点的计算任务。同时系统还将维护数据存储的可靠性，用多备份冗余存储机制提高数据存储的可靠性，并能及时检测和恢复出错的数据。

2、MapReduce的运行流程

2.1运行流程

由上图可以看到MapReduce执行下来主要包含这样几个步骤：

1) 首先正式提交作业代码，并对输入数据源进行切片

2) master调度worker执行map任务

3) worker当中的map任务读取输入源切片

4) worker执行map任务，将任务输出保存在本地

5) master调度worker执行reduce任务，reduce worker读取map任务的输出文件

6) 执行reduce任务，将任务输出保存到HDFS

2.2 运行流程详解

• 以WordCount为例

给定任意的HDFS的输入目录，其内部数据为“f a c d e……”等用空格字符分隔的字符串，通过使用MapReduce计算框架来统计以空格分隔的每个单词出现的频率，输出结果如,,形式的结果到HDFS目录中。

• WordCount运行图解

MapReduce将作业的整个运行过程分为两个阶段：Map阶段Reduce阶段。

Map阶段由一定数量的Map Task组成，流程如下：

• 输入数据格式解析：InputFormat
• 输入数据处理：Mapper
• 数据分区：Partitioner
• 数据按照key排序
• 本地规约：Combiner（相当于local reducer，可选）
• 将任务输出保存在本地

Reduce阶段由一定数量的Reduce Task组成，流程如下：

• 数据远程拷贝
• 数据按照key排序和文件合并merge
• 数据处理：Reducer
• 数据输出格式：OutputFormat

通常我们把从Mapper阶段输出数据到Reduce阶段的reduce计算之间的过程称之为shuffle。

MapReduce Java API应用

1、MapReduce开发流程

• 搭建开发环境，参考HDFS环境搭建，基本一致
• 基于MapReduce框架编写代码，Map、Reduce、Driver三部分组成。
• 编译打包，将源代码打成的包和依赖jar包打成一个包
• 上传至运行环境
• 运行hadoop jar命令，现已由yarn jar替代，建议使用新命令提交执行

具体提交命令为：

yarn jar testhdfs-jar-with-dependencies.jar com.tianliangedu.driver.WordCount /tmp/tianliangedu/input /tmp/tianliangedu/output3

• 通过yarn web ui查看执行过程
• 查看执行结果

2、WordCount代码实现

2.1 Map类编写

• Mapper：是MapReduce计算框架中Map过程的封装
• Text：Hadoop对Java String类的封装，适用于Hadoop对文本字符串的处理
• IntWritable：Hadoop对Java Integer类的封装，适用于Hadoop整型的处理
• Context：Hadoop环境基于上下文的操作对象，如Map中key/value的输出、分布式缓存数据、分布式参数传递等
• StringTokenizer：对String对象字符串的操作类，做基于空白字符的切分操作工具类
• 源码编写实现：

package com.tianliangedu.mapper;

import java.io.IOException;

import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

public class MyTokenizerMapper extends

Mapper {

// 暂存每个传过来的词频计数，均为1,省掉重复申请空间

private final static IntWritable one = new IntWritable(1);

// 暂存每个传过来的词的值，省掉重复申请空间

private Text word = new Text();

// 核心map方法的具体实现,逐个对去处理

public void map(Object key, Text value, Context context)

throws IOException, InterruptedException {

// 用每行的字符串值初始化StringTokenizer

StringTokenizer itr = new StringTokenizer(value.toString());

// 循环取得每个空白符分隔出来的每个元素

while (itr.hasMoreTokens()) {

// 将取得出的每个元素放到word Text对象中

word.set(itr.nextToken());

// 通过context对象，将map的输出逐个输出

context.write(word, one);

}

}

}

2.2 Reduce类编写

• Reducer：是MapReduce计算框架中Reduce过程的封装
• 源码编写实现：

package com.tianliangedu.reducer;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

//reduce类，实现reduce函数

public class IntSumReducer extends

Reducer {

private IntWritable result = new IntWritable();

//核心reduce方法的具体实现,逐个去处理

public void reduce(Text key, Iterable values,

Context context) throws IOException, InterruptedException {

//暂存每个key组中计算总和

int sum = 0;

//加强型for,依次获取迭代器中的每个元素值,即为一个一个的词频数值

for (IntWritable val : values) {

//将key组中的每个词频数值sum到一起

sum += val.get();

}

//将该key组sum完成的值放到result IntWritable中，使可以序列化输出

result.set(sum);

//将计算结果逐条输出

context.write(key, result);

}

}

2.3 Driver类编写

• Configuration：与HDFS中的Configuration一致，负责参数的加载和传递
• Job：作业，是对一轮MapReduce任务的抽象，即一个MapReduce的执行全过程的管理类
• FileInputFormat：指定输入数据的工具类，用于指定任务的输入数据路径
• FileOutputFormat：指定输出数据的工具类，用于指定任务的输出数据路径
• 源码编写实现：

package com.tianliangedu.driver;

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;

import com.tianliangedu.mapper.MyTokenizerMapper;

import com.tianliangedu.reducer.IntSumReducer;

public class WordCountDriver {

// 启动mr的driver方法

public static void main(String[] args) throws Exception {

// 得到集群配置参数

Configuration conf = new Configuration();

// 设置到本次的job实例中

Job job = Job.getInstance(conf, "天亮WordCount");

// 指定本次执行的主类是WordCount

job.setJarByClass(WordCountDriver.class);

// 指定map类

job.setMapperClass(MyTokenizerMapper.class);

// 指定combiner类，要么不指定，如果指定，一般与reducer类相同

job.setCombinerClass(IntSumReducer.class);

// 指定reducer类

job.setReducerClass(IntSumReducer.class);

// 指定job输出的key和value的类型,如果map和reduce输出类型不完全相同，需要重新设置map的output的key和value的class类型

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

// 指定输入数据的路径

FileInputFormat.addInputPath(job, new Path(args[0]));

// 指定输出路径,并要求该输出路径一定是不存在的

FileOutputFormat.setOutputPath(job, new Path(args[1]));

// 指定job执行模式，等待任务执行完成后，提交任务的客户端才会退出!

System.exit(job.waitForCompletion(true) ? 0 : 1);

}

}

2.4本地模拟分布式计算环境运行mapreduce

鉴于远程运行进行代码测试的复杂性，以及其它新框架均开始支持本地local环境模拟分布式计算运行， 故mapreduce从2.x开始也已经支持本地环境

具体做法请参见辅助资料集” 06-本地local环境模拟mapreduce并行计算的操作步骤”。

2.5 Maven打包

使用Maven命令，基于配置的Maven插件实现代码打包。

2.6 上传到运行环境

使用rz命令将打好的运行包上传到集群环境中。

2.7 运行WordCount程序

具体提交命令为：

yarn jar testhdfs-jar-with-dependencies.jar com.tianliangedu.driver.WordCount /tmp/tianliangedu/input /tmp/tianliangedu/output3

2.8 查看执行过程

Web访问地址为：http://cluster1.hadoop:8088/ui2/#/yarn-apps/apps

2.9 查看执行结果

3、标准代码实现

将之前的三个类，合并成一个类来处理

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;

//启动mr的driver类

public class WordCountDriver {

//map类，实现map函数

public static class MyTokenizerMapper extends

Mapper {

//暂存每个传过来的词频计数，均为1,省掉重复申请空间

private final static IntWritable one = new IntWritable(1);

//暂存每个传过来的词的值，省掉重复申请空间

private Text word = new Text();

//核心map方法的具体实现,逐个对去处理

public void map(Object key, Text value, Context context)

throws IOException, InterruptedException {

//用每行的字符串值初始化StringTokenizer

StringTokenizer itr = new StringTokenizer(value.toString());

//循环取得每个空白符分隔出来的每个元素

while (itr.hasMoreTokens()) {

//将取得出的每个元素放到word Text对象中

word.set(itr.nextToken());

//通过context对象，将map的输出逐个输出

context.write(word, one);

}

}

}

//reduce类，实现reduce函数

public static class IntSumReducer extends

Reducer {

private IntWritable result = new IntWritable();

//核心reduce方法的具体实现,逐个去处理

public void reduce(Text key, Iterable values,

Context context) throws IOException, InterruptedException {

//暂存每个key组中计算总和

int sum = 0;

//加强型for,依次获取迭代器中的每个元素值,即为一个一个的词频数值

for (IntWritable val : values) {

//将key组中的每个词频数值sum到一起

sum += val.get();

}

//将该key组sum完成的值放到result IntWritable中，使可以序列化输出

result.set(sum);

//将计算结果逐条输出

context.write(key, result);

}

}

//启动mr的driver方法

public static void main(String[] args) throws Exception {

//得到集群配置参数

Configuration conf = new Configuration();

//设置到本次的job实例中

Job job = Job.getInstance(conf, "天亮WordCount");

//通过指定相关字节码对象，找到所属的主jar包

job.setJarByClass(WordCountDriver.class);

//指定map类

job.setMapperClass(MyTokenizerMapper.class);

//指定combiner类，要么不指定，如果指定，一般与reducer类相同

job.setCombinerClass(IntSumReducer.class);

//指定reducer类

job.setReducerClass(IntSumReducer.class);

//指定job输出的key和value的类型,如果map和reduce输出类型不完全相同，需要重新设置map的output的key和value的class类型

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

//指定输入数据的路径

FileInputFormat.addInputPath(job, new Path(args[0]));

//指定输出路径,并要求该输出路径一定是不存在的

FileOutputFormat.setOutputPath(job, new Path(args[1]));

//指定job执行模式，等待任务执行完成后，提交任务的客户端才会退出!

System.exit(job.waitForCompletion(true) ? 0 : 1);

}

}

4、mapreduce作业运行的依赖项

• 提交job作业运行的linux帐户，必须在hdfs家目录中存在相应的家目录。

• 原因分析：
  • 作业运行过程中，会在当前linux帐户所属的hdfs家目录创建相应的临时文件夹和相应的临时文件，故必须有该先决条件。
• 操作步骤
  • 登陆帐户所在的linux主机当中，添加学员所属的班级组。
  • 将学员帐户，添加到该班级组当中。
  • 通过hdfs管理员帐户，为该学员创建hdfs家目录，并修改家目录的owner权限归该学员所有。
    • 添加组：groupadd job008
    • 改变用户所属组：usermod –G job008 ${user_name}
    • 用hdfs帐户在hdfs中创建家目录：hdfs dfs –mkdir /user/${user_name}
    • 将新创建的hdfs学员家目录的owner权限赋给所属学员：
      • Hdfs dfs –chown –R ${user_name} /user/${user_name}

MapReduce Shell应用

1、MapReduce的二级命令

mapred称为一级命令，直接输入mapred回车，即可查看二级命令：

2、MapReduce的三级命令

输入一级命令mapred后，再任意输入一个二级命令，即可查看三级命令：

3、MapReduce shell应用

• 查看当前正在执行的job任务

先提交一个WordCount任务，然后使用mapred job -list查看任务列表

• 终止(kill)一个任务的执行

由于某种原因，要立即终止某任务的执行，则使用mapred job -kill job-id。

构造场景：先提交一个WordCount job，然后通过kill job-id来终止任务：

• 查看一个job的日志

使用mapred shell命令，通过job-id可以查看job的工作日志。

命令格式为：mapred job -logs job-id：

MapReduce技术特征

1、向“外”横向扩展，而非向“上”纵向扩展

• 集群的构建完全选用价格便宜、易于扩展的低端商用服务器，而非价格昂贵不易扩展的商用服务
• 大规模数据处理和大规模数据存储的需要，讲求集群综合能力，而非单台机器处理能力，横向增加机器节点数据量

2、失效被认为是常态

• 使用大量普通服务器，节点硬件和软件出错是常态
• 具备多种有效的错误检测和恢复机制，在某个计算节点失效后会自动转移到别的计算节点。某个任务节点失败后其他节点能够无缝接管失效节点的计算任务
• 当失效节点恢复后自动无缝加入集群，不需要管理员人工进行系统配置

3、移动计算，把处理向数据迁移(数据本地性)

• 采用代码/数据互定位的功能，计算和数据在同一个机器节点或者是同一个机架中，发挥数据本地化特点
• 可避免跨机器节点或是机架传输数据，提高运行效率

4、顺序处理数据、避免随机访问数据

• 磁盘的顺序访问远比随机访问快得多，因此MapReduce设计为面向顺序式大规模数据的磁盘访问处理
• 利用集群中的大量数据存储节点同时访问数据，实现面向大数据集批处理的高吞吐量的并行处理

5、推测执行

• 一个作业由若干个Map任务和Reduce任务构成，整个作业完成的时间取决于最慢的任务的完成时间。由于节点硬件、软件问题，某些任务可能运行很慢
• 采用推测执行机制，发现某个任务的运行速度远低于任务平均速度，会为慢的任务启动一个备份任务，同时运行。哪个先运行完，采用哪个结果。

6、平滑无缝的可扩展性

• 可弹性的增加或减少集群计算节点来调节计算能力
• 计算的性能随着节点数的增加保持接近线性程度的增长

7、为应用开发隐藏系统底层细节

• 并行编程有很多困难，需要考虑多线程中复杂繁琐的细节，诸如分布式存储管理、数据分发、数据通信和同步、计算结果收集等细节问题。
• MapReduce提供了一种抽象机制将程序员与系统层细节隔离开，程序员只需关注业务，其他具体执行交由框架处理即可。

MapReduce项目练习

1、在hdfs目录/tmp/tianliangedu/input/wordcount中有一系列文件，内容均为","号分隔，

求按","号分隔的各个元素的出现频率，输出到目录/tmp/tianliangedu/output/个人用户名的hdfs目录中。（必做）

2、在hdfs目录/tmp/tianliangedu/input/wordcount目录中有一系列文件，内容为","号分隔，分隔后的元素均为数值类型、字母、中文，求所有出现的数值的和。（必做）

3、在hdfs目录/tmp/tianliangedu/input/wordcount目录中有一系列文件，内容为","号分隔，分隔后的元素均为数值类型、字母、中文，求所有出现的数值的平均值。（必做）

4、在hdfs目录/tmp/tianliangedu/input/wordcount目录中有一系列文件，内容为","号分隔，分隔后的元素均为数值类型、字母、中文，求数值类型、字母类型、中文类型各自的次数。（必做）

5、在hdfs目录/tmp/tianliangedu/input/wordcount目录中有一系列文件，内容为","号分隔，同时在hdfs路径/tmp/tianliangedu/black.txt黑名单文件，一行一个单词用于存放不记入统计的单词列表。求按","号分隔的各个元素去除掉黑名单后的出现频率，输出到目录/tmp/tianliangedu/output/个人用户名的hdfs目录中。 （必做）

6、在hdfs目录/tmp/tianliangedu/input/wordcount目录中有一系列文件，求这些文件一共有多少行？（类似于mysql数据库中的select count(*) from table)。（必做）

7、在hdfs目录/tmp/table/student中存在student.txt文件，hdfs目录/tmp/table/student_location中存在student_location.txt文件，求两个hdfs目录中共有多少行？

8、在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），hdfs目录/tmp/table/student_location中存在student_location.txt文件，按tab分隔，字段名为（学号，省份，城市，区名），在Map任务中用student_location.txt文件中的学号过滤student.txt中的学号字段，输出student.txt中的存在交集的记录，输出结果结构按tab分隔后的四个字段为（学号，姓名，课程号，班级名称）。

9、（列筛选不去重) 在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），选择学号和班级名称列不去重输出，输出结果结构按tab分隔后的两个字段为（学号，班级名称）。

10、 (列筛选去重)在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），选择学号和班级名称列去掉重复存在行输出，输出结果结构按tab分隔后的两个字段为（学号，班级名称）。

11、 (列值replace操作)在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），将班级名称为"计算机*班"的更换成"计算机科学与技术*班",不做去重，输出结果结构按tab分隔后的两个字段为（学号，班级名称）。

12、 (列分组统计操作)在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），将课程号为分组，输出结果结构按tab分隔后的两个字段为（课程号，分组人数）。

13、 (列分组统计操作)在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），将班级名称为分组，输出结果结构按tab分隔后的两个字段为（班级名称，分组人数）。

14、 (多列分组统计操作)在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），将课程号和班级名称为各自分组，输出结果结构按tab分隔后的两个字段为（分组名称，分组人数），即如(课程号，分组人数)、(班级名称，分组人数)。

15、 (两表inner join) 在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），hdfs目录/tmp/table/student_location中存在student_location.txt文件，按tab分隔，字段名为（学号，省份，城市，区名），对两个hdfs目录的按学号求交集，输出结果结构按tab分隔后的四个字段为（学号，姓名，课程号，班级名称）。

16、 (两表left join) 在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），hdfs目录/tmp/table/student_location中存在student_location.txt文件，按tab分隔，字段名为（学号，省份，城市，区名），对两个hdfs目录的按学号以student为主表，求左外链接操作，输出结果结构按tab分隔后的七个字段为（学号，姓名，课程号，班级名称，省份，城市，区名）。

17、 (两表right join) 在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），hdfs目录/tmp/table/student_location中存在student_location.txt文件，按tab分隔，字段名为（学号，省份，城市，区名），对两个hdfs目录的按学号以student为主表，求右外链接操作，输出结果结构按tab分隔后的七个字段为（学号，姓名，课程号，班级名称，省份，城市，区名）。

18、 (两表full join) 在hdfs目录/tmp/table/student中存在student.txt文件，按tab分隔，字段名为(学号，姓名，课程号，班级名称），hdfs目录/tmp/table/student_location中存在student_location.txt文件，按tab分隔，字段名为（学号，省份，城市，区名），对两个hdfs目录的按学号以student为主表，求全外链接操作，输出结果结构按tab分隔后的七个字段为（学号，姓名，课程号，班级名称，省份，城市，区名）。