BigData CH4 MapReduce

CH4 MapReduce 并行编程框架

并行计算

划分子任务、计算、合并结果

构抽象模型Map与Reduce

Map

对一组数据元素进行某种重复式的处理

map: (k1; v1) → \rightarrow → [(k2; v2)]

输入：键值对(k1; v1)表示的数据

处理：文档数据记录(如文本文件中的行，或数据表格中的行)将以“键值对”形式传入map函数；map函数将处理这些键值对，并以另一种键值对形式输出处理的一组键值对中间结果[(k2; v2)]

输出：键值对[(k2; v2)]表示的一组中间数据

Reduce

对Map的中间结果进行某种进一步的结果整理

例如关系数据库的聚合函数

reduce: (k2; [v2]) → \rightarrow → [(k3; v3)]

输入：由map输出的一组键值对[(k2; v2)] 将被进行合并处理将同样主键下的不同数值合并到一个列表[v2]中，故reduce的输入为(k2; [v2])

处理：对传入的中间结果列表数据进行某种整理或进一步的处理,并产生最终的某种形式的结果输出[(k3; v3)] 。

输出：最终输出结果[(k3; v3)]

Reduce前，必须所有map都做完，因此需要一个同步障barrier

这个阶段也负责对map中间结果数据进行收集整理和处理(aggregation、shuffle)

编程模型和框架

wordcount 伪代码

Combiner和Partitioner
- Combiner
  
  在每个map节点输出中间结果的键值对前，进行合并处理，
  
  例如两个(good,1) 合并成(good,2)
  
  执行是在Map节点完成计算、输出中间结果之前
- Partitioner
  
  分区处理
  
  目的：消除数据传入Reduce结点之后带来的不必要的相关性。
  
  在Map到Reduce中间数据整理阶段完成
  
  例如：保证所有主键相同的<k,v>都输入到同一个Reduce节点，避免Reduce再访问其他Reduce
完整的MapReduce并行编程框架

MapReduce提供了统一的计算框架可以完成
- 计算任务的划分和调度
- 数据分布存储和划分
- 处理数据和计算任务同步
- 结果数据的收集整理：sorting、combining、partitioning
- 系统通信、负载平衡、计算性能优化处理
- 节点出错检测、失效恢复
MapReduce提供的主要功能
1. 任务调度：提交的一个计算作业(job)将被划分为很多个计算任务(tasks), 任务调度功能主要负责为这些划分后的计算任务分配和调度计算节点(map节点或reducer节点); 同时负责监控这些节点的执行状态, 并负责map节点执行的同步控制(barrier); 也负责进行一些计算性能优化处理, 如对最慢的计算任务采用多备份执行、选最快完成者作为结果
2. 数据/代码互定位：为了减少数据通信，一个基本原则是本地化数据处理(locality)，即一个计算节点尽可能处理其本地磁盘上所分布存储的数据，这实现了代码向数据的迁移；当无法进行这种本地化数据处理时，再寻找其它可用节点并将数据从网络上传送给该节点(数据向代码迁移)，但将尽可能从数据所在的本地机架上寻找可用节点以减少通信延迟
3. 出错处理：以低端商用服务器构成的大规模MapReduce计算集群中,节点硬件(主机、磁盘、内存等)出错和软件有bug是常态，因此,MapReducer需要能检测并隔离出错节点，并调度分配新的节点接管出错节点的计算任务
4. 分布式数据存储与文件管理：海量数据处理需要一个良好的分布数据存储和文件管理系统支撑,该文件系统能够把海量数据分布存储在各个节点的本地磁盘上,但保持整个数据在逻辑上成为一个完整的数据文件；为了提供数据存储容错机制,该文件系统还要提供数据块的多备份存储管理能力
MapReduce的主要设计思想与特点
- 向外扩展而非向上纵向扩展
  - 即MapReduce集群的构筑选用价格便宜、易于扩展的大量低端商用服务器，而非价格昂贵、不易扩展的高端服务器（SMP）
- 失效被认为是常态
  - 一个良好设计、具有容错性的并行计算系统不能因为节点失效而影响计算服务的质量，任何节点失效都不应当导致结果的不一致或不确定性；任何一个节点失效时，其它节点要能够无缝接管失效节点的计算任务；当失效节点恢复后应能自动无缝加入集群，而不需要管理员人工进行系统配置、
- 把处理向数据迁移
  - 计算节点将首先将尽量负责计算其本地存储的数据,以发挥数据本地化特点(locality),仅当节点无法处理本地数据时，再采用就近原则寻找其它可用计算节点，并把数据传送到该可用计算节点
- 顺序处理数据、避免随机访问数据
- 为应用开发者隐藏系统层细节
- 平滑无缝的可扩展性

基本工作原理

数据存储与计算节点

基本工作过程

在client node的client JVM里面提交MapReduce程序，然后run job on JobClient，Client需要复制Job资源到HDFS，然后get new job ID，之后才正式提交Job到JobTracker。
JobTracker 会初始化job，然后进行查询输入分片
TaskTracker 会提交heartheat给JobTracker
TaskTracker 会 launch 一个child JVM 进行运行 MapTask 或者 ReduceTask

TaskTracker 与 Datanode 一对一，一般是看数据在哪一节点，就在哪个节点启动对应的TaskTracker。

指定输入格式，inputformat
然后进行split
RecordReaders 读取每个记录转化成<k,v>
输入到Map Task
经过combiner、partition、同步障
然后进行sort
reduce
输出(k,v)
最后写回HDFS

主要组件以及编程接口

输入格式InputFormat

InputFormat提供了以下一些功能
- 选择文件或者其它对象，用来作为输入
- 定义InputSplits, 将一个文件分为不同任务
- 为RecordReader提供一个基础，用来读取这个文件
有一个抽象的类FileInputFormat，所有的输入格式类都从这个类继承其功能以及特性。当启动一个Hadoop任务的时候，一个输入文件所在的目录被输入到FileInputFormat对象中。FileInputFormat从这个目录中读取所有文件。然后FileInputFormat将这些文件分割为多个InputSplits
格式
- TextInput Format
- KeyValueTextInput Format
- SequenceFileInput Format
一个MapReduce程序被统称为一个Job，可能有上百个任务构成

RecordReader

RecordReader实际上定义了如何将数据记录转化为一个(key,value)对的详细方法，并将数据记录传给Mapper类

Mapper

每一个Mapper类的实例生成了一个Java进程，负责处理某一个InputSplit上的数据。

Combiner

合并相同key的键值对，减少partitioning时候的数据通信开销。用户可以使用JobConf.setCombinerClass(class)自定制Combiner

Partition and Shuffle

在Map完成之后，每一个Map函数的结果传到对应Reducer所在的节点，此时，可以提供一个Partition类，用来决定一个给定的(key,value)传给哪个Reduce节点

Sort

传到Reducer节点上的K,V会被Hadoop自动排序(即Map生成的结果传送到某个节点的时候会被自动排序)

Reducer

做用户定义的Reduce操作，OutputFormat输出

OutputFormat

继承自FileOutputFormat，每一个Reducer都写一个文件到一个共同的输出目录，名字是part-nnnn

FileOutputFormat.setOutputPath();
JobConf.setOutputFormat();

容错处理

失败的任务再次执行

TaskTracker会把状态信息汇报给JobTracker，最终由JobTracker决定重新执行哪个任务

投机执行：为了加快执行速度，Hadoop会自动重复执行同一个任务，以最先执行成功的为准。

可解决的算法问题

搜索引擎：文档倒排索引、网页链接图分析、页面排序

Web日志分析、文档分析处理、机器学习、机器翻译等

MapReduce排序算法

外排序与MapReduce的Sort

构建单词同现矩阵算法

word co-occurrence matrix

是一个二维的N*N矩阵
N是词汇量(不同单词的数目)
矩阵元素M(i,j) 标识单词Wi 和 Wj 在一定范围内同时出现的次数(可以定义窗口大小)

空间开销 O ( n 2 ) O(n^2) O(n2) ，内存和磁盘之间的换页会导致执行十分缓慢

M.R. 算法伪码

Mapper：文档d的每个单词，如果是同现窗口(word.neighbor)，对这个pair count 一个 1
shuffle & sort 把相同的pair 的count 变成一个count数组
reduce: sum counts[i] -> s

专利文献数据分析

做的事是专利被引

最后输出: cited ; value: citing id1, citing id2…