大数据及MapReduce编程模型94

唐卓博ust_tz@126.com湖南大学信息科学与工程学院2014年8月大数据及其并行编程模型概述主要内容一、大数据概述二、应对大数据的系统思维三、MapReduce并行编程详解2注：本课件前30页PPT来源于国防科大李东升教授：“大数据时代的挑战和探索”互联网应用数据急剧增长

互联网用户数量巨大，日益活跃

•

微博、论坛、电子商务网站等

•

互联网上的用户生成数据（User

Generated

Content，

UGC）淘宝网每天新增数据40TB以上百度每天处理10PB量级的数据，总数据量达1000PB应用背景注：本课件前30页PPT来源于国防科大李东升教授：“大数据时代的挑战和探索”一、大数据概述•

随着信息化的推进，国民经济、国家安全

等领域的数据不断增长

物联网、移动通信电话、手机短信、语音数据

遥感、公共安全、医疗、交通、情报等很多领域

•

高分辨率卫星（影像）、城市监控摄像头（视频）、…

•

据报道，武汉监控摄像头已超过25万个，如采用1080P高清摄

像头（一天产生数据量40GB以上），整个城市每天新增监控

数据10PB以上应用背景•

科学实验数据规模巨大，增长迅猛生物工程气候监测高能物理天文观测生态环境

….气候研究华大基因测序目前每天产生数据约15TB，一年超过5PB

一欧洲CERN对撞机每年产生的数据量超过15

PB基因测序应用背景全球数据量•IDC报告预测：未来

十年，全球数据量继

续迅速增长Amount

of

digital

informationcreated

and

replicated

in

a

year––––年均增长率超过40%2009年0.8ZB2020年35ZB1ZB~106PB月球容量4GB的DVD光用容量4GB的DVD光盘存储，DVD可从地球排至月球G-T-P-E-Z-Y•

维基（Wiki）百科的定义

Big

data

is

a

collection

of

data

sets

so

large

and

complex

that

it

becomes

difficult

to

process

using

on-hand

database

management

tools

•

IDC的定义

Big

data

technologies

describe

a

new

generation

of

technologies

and

architectures,

designed

to

economically

extract

value

from

very

large

volumes

of

a

wide

variety

of

data,

by

enabling

high-velocity

capture,

discovery,

and/or

analysis.

什么是大数据大数据是超大、复杂的数据集，现有的数据库管理技术难以应对大数据技术描述了新一代的技术和架构，通过高速的数据获取、发现和分析技术，以经济的方式从各种超大规模的数据中提取价值一、大数据概述•

Volume：规模大

从PB级到ZB级

1

ZB

~

106*

PB•

Variety：多样化

结构化、非结构化

文本、图像、视频等•

Velocity：变化快

批处理/离线数据、流/实时/在线数据等•

Value/

Veracity：价值稀疏

/数据质量

噪音和无用信息很多一、大数据概述大数据的特点•

大数据技术对经济社会和科研都在产生重

要影响

–

互联网产业、电子商务推荐、日常生活

大数据的影响季节性流感是一个重要的公共卫生问题：WHO估计，全球每年25万至50万人因此死亡及时监测到疾病的传播情况，尽快采取应对措施2008年，Google通过处理网络搜索日志中的几千亿查询数据，训练建立流感疾病监测的数学模型，比美国病控制和预防中心提前1-2周给出流感的传播情况论文发表在Nature(2009.2):DetectingInfluenza

EpidemicsusingSearchEngineQueryData•

大数据技术对经济社会和科研都在产生重

要影响

–

科学研究

三种科研模式：理论、实验、计算第四模式：数据密集型的科学发现图灵奖获得者JimGray2007年提出专辑：Nature(2008.9):”Big

Data”,Science(2011.2):”Dealing

with

data”大数据的影响•

2012年3月29日，美国政府宣布投资2亿

美元启动“大数据研发计划”（

Big

Data

R&D

Initiative

）

美NSF、国防部、能源部、卫生总署等七部委•

我国科技部和基金委等部门高度重视

2013年973新立项项目：2项

“十二五”

国家科技计划信息技术领域2013年度备选项

目征集指南•

国内外学术界的热点课题

SIGMOD、

VLDB、OSDI、NSDI等著名会议

Nature、Science杂志11大数据成为热点课题•

传统技术难以应对大数据的规模

–

数据存储及访问的挑战当前较快硬盘的传输速度6Gbps，线性扫描10PB数据，需约19天而百度、Google等互联网公司每天处理

的数据量超过10PB案例源于：北航/爱丁堡樊文飞教授

•

可扩展是大规模分布式系统面临的基础性问题

–Jim

Gray（图灵奖获得者）将可扩展问题列为信

息技术领域需解决的16个长远问题之首Jim

Gray.

What

Next?

A

Few

Remaining

Problems

in

Information

Technology.

ACM

Turing

Award

Lecture

(1999).

Available

at

http:///enus/um/people/gray/talks/Gray_Turing_FCRC.ppt大数据带来的挑战（1）•

很多大数据应用对响应时间要求高

–

规模大、响应快：对存储和处理提出了很大挑战

–2007年前,Facebook使用数据库，总数据量15TB

•

目前，Facebook每天新增加的数据约70TB

–

传统并行数据库扩展性受限，节点规模很少超过100，

且价格昂贵

•2011年，Facebook系统具有2700多个节点，Google单个数据中心在上

万个节点集群上存储了约10PB数据•

如何设计可扩展、低成本、快速响应的大

数据存储和处理系统？大数据存储与处理的可扩展难题数据种类多，需求多样，关联复杂

–文本、图像、图形、视频等

–在线/流数据、离线/批处理等如何建模、存储、查询、分析和理解多样

化的复杂数据，挖掘数据价值？

–

大数据中垃圾和珍宝并存

–大海捞针、去粗取精、去伪存真

–需要计算机专家和领域专家的配合….大数据面临的挑战（2）传统算法在大数据时代可能不再有效

多项式时间算法O(Nk)，N太大

需要计算复杂性和算法设计理论上的变革

需要大数据计算思维上的变化

例如，从确定性计算到非精确性计算

商品在线推荐：只需要计算出前10名相关的结果，有

一点不准确也没有关系传统算法结论在大数据时代需要重新评估

–

简单方法+大数据集可能取得很好的结果大数据面临的挑战（3）•

2007年,Google公司的Brants等人研究了机

器翻译领域中基于单词训练数据集的语言

模型

比较了当时最先进的KN算法

与其提出的一个简单算法SB

研究表明，简单算法在小数

据集时效果不佳，但在大数

据集时，简单算法却产生了

更好的效果

T.Brants,A.C.Popat,etal.LargeLanguageModelsinMachineTranslation.

ProceedingsoftheJointConferenceonEmpiricalMethodsinNatural

LanguageProcessingandComputationalNaturalLanguageLearning,2007.16传统算法结论需要重新评估•

大数据时代的算法新理论

–

新的计算复杂性和算法设计理论•

复杂大数据的建模、表示和可视化

–

多源异构大数据：由大到小•

面向大数据的新型存储和计算系统架构

–大规模并行/分布处理•

大数据（并行）挖掘算法及应用大数据的研究课题主要内容一、大数据概述二、应对大数据的系统思维三、MapReduce并行编程详解2181.

数据为中心的计算架构计算和存储唇齿相依2.化繁为简，分而治之

可扩展的数据并行处理3.求同存异，聚焦领域放松传统数据处理技术的约束，如一致性等、行式存储-列式存储高可扩展高吞吐率高可靠性……主要内容18二、应对大数据的系统思维1.

数据为中心的计算架构过去20年来,计算器件的带宽提升了100–2000倍，而延迟改善只有5-20倍CPU

和

on-chip

L2之间:

带宽:增长了2250倍

延迟:降低了20倍L3

cache

和DRAM之间：

带宽:增长了125倍

延迟:降低了4倍DRAM

和disk之间:

带宽:增长了150倍

延迟:降低了8倍

LAN连接的两个节点之间

:

带宽:增长了100倍

延迟:降低了15倍充分利用数据和存储的局部性（缓存、复制、预取）延迟提升滞后于带宽Source：CACM（Patterson)二、应对大数据的系统思维1.

数据为中心的计算架构（续）20二、应对大数据的计算思维

–

数据分布存储在计算附近？–

计算尽量利用数据局部性–

存储架构、互连网络架构数据密集型计算计算密集型计算

SystemData–

数据存储与计算相分离–

计算之前加载数据–

规模挑战：元数据管理+数

据传输221.

数据为中心的计算架构（续）案例：MicrosoftFlatDatacenterStorage(OSDI2012)MinuteSort新架构+高效互连网络•

简化的可扩展数据并行处理：MapReduce框架Map:

Key1/Value1

（输入数据）Reduce:

Key2/Value2

（中间数据）Key2/Value2

（中间数据）

Value

（输出数据）数据按照key进行分区：数据并行Google提出（OSDI’04）

中间数据输出数据输入数据222.

化繁为简，分而治之•

特点

每个Map/Reduce任务相对独立，执行的任务简单

简单，易于扩展（应用无需修改）、容错性好（复算）

缺点：Map和Reduce阶段之间需要大量的数据交换•

开源实现

Hadoop及其变型

成功应用于众多著名公司

•

Facebook,

Yahoo!

,

AOL,

EBay,

IBM,

….

•

百度，阿里巴巴等MapReduce数据并行框架•

MapReduce

革新

MapReduce

Online

(UC

Berkeley)、

HadoopDB

(

(Yale)

)

Hadoop++

(Dittrich

et

al.:

VLDB’2012)

Spark（内存Hadoop，

UCBerkeley）…•

新的数据并行处理框架

Pregel，GraphLab:

图数据的并行处理框架

Dremel：

快速交互式数据分析系统，PB/s

Storm：流处理数据框架

….学术界和工业界不懈努力数据一致性

关系数据库：强一致性Atomicity

Consistency

Isolation

Durability（

ACID）写操作完成后，任何后续读操作将得到最新值•

分布式环境下，强一致性的代价昂贵，很

多应用也无需强一致性弱化数据一致性，提升可扩展性和可靠性3.

求同存异，聚焦领域Youcanhaveatmosttwoofthesepropertiesforanyshared-datasystem在分布式系统中，数据一致性、系统可用性、以及对网络断分容忍性中，任何时候只能实现其中两个特性UC

Berkeley的Eric

Brewer提出猜想（2000）MIT的Nancy

Lynch等予以证明（2002）

CAP定理

为什么牺牲数据一致性？

牺牲P、A对互联网上的大数据

应用来说难以容忍牺牲C的代价可以接受应用开发稍显复杂很多应用并不关心C弱（最终）一致性Basically

Available

Soft-stateEventual

Consistency（Base）弱化数据一致性•

聚焦领域应用需求，简（优）化系统设计

GFS：聚焦于数据“读多写少”场景

满足可扩展性、可用性等多种需求的平衡•

NoSQL

存储

很多领域应用只需要对数据进行简单的读写

•

不需要复杂的SQL操作，如skyline查询、多表join等key/value存储放弃SQL的某些要求列式存储283.

求同存异，聚焦领域

全能选手

Vs.

特长生

大数据研究正方兴未艾•

Gartner：Hype

Cycle

2012主要内容一、大数据概述二、应对大数据的系统思维三、MapReduce并行编程详解2MapReduce起源：Google搜索每一次搜索200+CPU200TB以上数据1010CPU周期0.1秒内响应5¢广告收入计算问题简单，但求解困难待处理数据量巨大（PB级），只有分布在成百上千个节点上并行计算才能在可接受的时间内完成如何进行并行分布式计算？如何分发待处理数据？如何处理分布式计算中的错误？简单的问题，计算并不简单！MapReduce：大规模数据处理处理海量数据（>1TB）上百/上千CPU实现并行处理简单地实现以上目的"GoogleEarthuses70.5TB:70TBfortherawimageryand500GBfortheindexdata."From:/2006/09/how-much-data-does-google-store.html分而治之DivideandConquer

GoogleMapReduce架构设计师JeffreyDeanMapReduce特性自动实现分布式并行计算容错提供状态监控工具模型抽象简洁，程序员易用MapReduce特性MapReduce程序是设计用来并行计算大规模海量数据的，这需要把工作流分划到大量的机器上去，如果组件(component)之间可以任意的共享数据，那这个模型就无法扩展到大规模集群上去（数百或数千个节点），用来保持节点间数据的同步而产生的通信开销会使得系统在大规模集群上变得不可靠和效率低下所有在MapReduce上的数据元素都是不可变的，这就意味着它们不能够被更新。如果在一个mapping任务中你改变了一个输入键值对，它并不会反馈到输入文件；节点间的通信只在产生新的输出键值对((key,value)pairs)时发生，Hadoop系统会把这些输出传到下一个执行阶段。MapReducemapping和reducing函数接收数值（键，值）对mapper可能把一个输入map为0个，1个或100个输出reducer可能计算超过一个的输入列表并生成一个或多个不同的输出MapReduce编程模型用户只需要实现两个函数接口：map(in_key,in_value)-> (out_key,intermediate_valuelist)reduce(out_key,intermediate_valuelist)->out_valuelist输入的key和value的类型和输出的类型可以是不同的map将数据源中的记录（文本中的行、数据库中条目等）作为map函数中的key*value对例如（filename,line）map()将生成一个或多个中间结果，以及与input相对应的一个outputkeyreducemap操作结束后，所有与某指定outkey相对应的中间结果组合为一个列表（list）。reduce()函数将这些中间结果组合为一个或多个对应于同一outputkey的finalvalue每一个outputkey通常只有一个finalvaluereduce()个数可以为0个或多个任务执行过程源文件：GFSMap处理结果：本地存储Reduce处理结果：GFS日志：GFS文件存储位置Shuffle和Sort当Map开始产生输出时，并不是简单的把数据写到磁盘，因为频繁的磁盘操作会导致性能严重下降。它的处理过程更复杂，数据首先是写到内存中的一个缓冲区，并进行预排序，以提升效率。CombinerCombiner并行化map()函数可以并行执行，为不同的输入数据集生成不同的中间结果reduce()函数也可以并行执行，分别处理不同的outputkeymap和reduce的处理过程中不发生通信瓶颈：只有当map处理全部结束后，reduce过程才能够开始MapReduce的并行执行Worker故障Master周期性的ping每个worker。如果master在一个确定的时间段内没有收到worker返回的信息，那么它将把这个worker标记成失效重新执行该节点上已经执行或尚未执行的Map任务重新执行该节点上未完成的Reduce任务，已完成的不再执行Master故障定期写入检查点数据从检查点恢复MapReduce的容错任务备份机制慢的workers会严重地拖延整个执行完成的时间由于其他的任务占用了资源磁盘损坏解决方案：推测性的执行（Speculativeexecution）在即将完成时，备份任务多个worker同时进行相同的任务任何一个完成均可可以十分显著地提高执行效率MapReduce的优化本地处理Master调度策略:向GFS询问获得输入文件blocks副本的位置信息Maptasks的输入数据通常按64MB来划分(GFSblock大小)按照blocks所在的机器或机器所在机架的范围进行调度效果绝大部分机器从本地读取文件作为输入，节省大量带宽MapReduce的优化跳过有问题的记录一些特定的输入数据常导致Map/Reduce无法运行调试或者修改在每个worker里运行一个信号处理程序，捕获map或reduce任务崩溃时发出的信号，一旦捕获，就会向master报告，同时报告输入记录的编号信息。如果master看到一条记录有两次崩溃信息，那么就会对该记录进行标记，下次运行的时候，跳过该记录MapReduce的优化MapReduce示例：单词计数案例：单词记数问题(WordCount)给定一个巨大的文本（如1TB），如何计算单词出现的数目？MapReduce示例：单词计数使用MapReduce求解该问题Step1:自动对文本进行分割MapReduce示例：单词计数使用MapReduce求解该问题Step2:在分割之后的每一对<key,value>进行用户定义的Map进行处理，再生成新的<key,value>对MapReduce示例：单词计数使用MapReduce求解该问题Step3:对输出的结果集归拢(不同mapslot间copy到一起)、排序(sort)(系统自动完成）MapReduce示例：单词计数使用MapReduce求解该问题Step4:通过Reduce操作生成最后结果MapReduce示例：单词计数使用MapReduce求解该问题定义Map和Reduce函数map(Stringinput_key,Stringinput_value)://input_key:documentname//input_value:documentcontents

foreachwordwininput_value:

EmitIntermediate(w,"1");reduce(Stringoutput_key,Iteratorintermediate_values)://output_key:aword//output_values:alistofcounts

intresult=0;

foreachvinintermediate_values:result+=ParseInt(v);

Emit(AsString(result));其他示例分布式检索map函数挑选出满足特定模式的行，并将其组装成元组输出。reduce函数是一个简单的确认函数，它完成的工作仅仅是将中间元组拷贝到输出中。计算URL访问频率map函数处理web网页的访问日志，并输出<URL,1>。reduce函数将每个URL的访问次数加起来，输出<URL,totalcount>其他示例翻转web-link图在每个作为源的页面中，检查其连接URL，并逐个输出<target,source>元组。reduce函数将连接到每个target的所有source组合起来，形成list列表，输出<target,list(source)>每个站点的术语向量术语向量表示出在一篇文章中或者一组文章中最重要的单词，通常以<word,frequency>元组的方式。map函数输出每个文章的<hostname,termvector>（hostname通过文章的URL分析得到）。reduce函数取出不常用的术语，将其余的相加，得到最终的<hostname,termvector>对其他示例倒排索引map函数分析每个文档,然后产生一个(词,文档号)对的序列.reduce函数接受一个给定词的所有对,排序相应的文档IDs,并且产生一个(词,文档ID列表)对.所有的输出对集形成一个简单的倒排索引分布式排序map函数从每个记录提取key,并且产生一个(key,record)对.reduce函数不改变任何的对.“实践是检验真理的唯一标准”实践证明，MapReduce是出色的分布式计算模型Google宣布，其对分布于1000台计算机上的1TB数据进行排序仅仅需要68s对4000台计算机上的1PB数据进行排序处理仅需要6小时2分钟（每次测试至少会损坏1块硬盘）在08年1月份，GoogleMapReduce平均每天的数据处理量是20PB，相当于美国国会图书馆当年5月份存档网络数据的240倍Hadoop上的MapReducejob:是客户端程序想要完成的一系列工作的集合。包括输入数据，MapReduce程序和配置信息。task:Hadoop将job分解为tasks有两种类型的task:maptask和reducetaskjobtracker和tasktracker:用来控制job执行的tasktracker运行task，并向jobtracker报告进度信息jobtracker记录下每一个job的进度信息，如果一个task失败，jobtracker会将其重新调度到另外的tasktracker上。Hadoop-MapReduce工作原理HadoopStreamingandPipesHadoop流允许用Java以外的语言来编写Map和Reduce函数Hadoop管道C++接口流和管道及子进程的关系进度和状态更新Hadoop-MapReduceMapperpublicstaticclass**MapperextendsMapper<Object,Text,Text,IntWritable>Reducer

publicstaticclass**Reducerextendseducer<Text,IntWritable,Text,IntWritable>DriverPackage

org.apache.hadoop.mapreduce.Job; org.apache.hadoop.mapreduce.Mapper; org.apache.hadoop.mapreduce.Reducer;接口描述publicinterfaceMapper<K1,V1,K2,V2>extendsJobConfigurable,Closeable{voidmap(K1key,V1value,OutputCollector<K2,V2>output,Reporterreporter)throwsIOException;}publicinterfaceReducer<K2,V2,K3,V3>extendsJobConfigurable,Closeable{voidreduce(K2key,Iterator<V2>values,OutputCollector<K3,V3>output,Reporterreporter)throwsIOException;}MapperpublicstaticclassMapextendsMapper<LongWritable,Text,Text,IntWritable>{privatefinalstaticIntWritableone=newIntWritable(1);privateTextword=newText();publicvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{Stringline=value.toString();StringTokenizertokenizer=newStringTokenizer(line);while(tokenizer.hasMoreTokens()){word.set(tokenizer.nextToken());context.write(word,one);}}}ReducerpublicstaticclassReduceextendsReducer<Text,IntWritable,Text,IntWritable>{publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intsum=0;for(IntWritableval:values){ sum+=val.get();}context.write(key,newIntWritable(sum));}}Driverpublicstaticvoidmain(String[]args)throwsException{Configurationconf=newConfiguration();Jobjob=newJob(conf,"wordcount");job.setJarByClass(WordCount.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);job.setMapperClass(Map.class);job.setReducerClass(Reduce.class);job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);FileInputFormat.addInputPath(job,newPath(args[0]));FileOutputFormat.setOutputPath(job,newPath(args[1]));job.waitForCompletion(true);}InputFiles输入文件一般保存在HDFS中文件的类型不固定，可能是文本的，也有可能是其它形式的文件文件经常很大，甚至有几十个GBInput会被分成inputsplit，split由record组成。map处理每一个record，并且返回key和value的对MapReduce程序并不需要直接处理InputSplit，由InputFormat创建的InputSplitsInputSplit定义了输入到单个Map任务的输入数据InputSplit将文件分为64MB的大小hadoop-site.xml中的mapred.min.split.size参数控制这个大小mapred.tasktracker.map.taks.maximum用来控制某一个节点上所有map任务的最大数目RecordReaderInputSplit定义了一项工作的大小，但是没有定义如何读取数据RecordReader实际上定义了如何从数据上转化为一个(key,value)对，从而输出到Mapper类中TextInputFormat提供了LineRecordReaderInputFormat定义了这些文件如何分割，读取InputFile提供了以下一些功能选择文件或者其它对象，用来作为输入定义InputSplits，将一个文件分开成为任务为RecordReader提供一个工厂，用来读取这个文件有一个抽象的类FileInputFormat，所有的输入格式类都从这个类继承这个类的功能以及特性。当启动一个Hadoop任务的时候，一个输入文件所在的目录被输入到FileInputFormat对象中。FileInputFormat从这个目录中读取所有文件。然后FileInputFormat将这些文件分割为一个或者多个InputSplits。通过在JobConf对象上设置JobConf.setInputFormat设置文件输入的格式预定义的文件输入格式InputFormat:Description:Key:Value:TextInputFormatDefaultformat;readslinesoftextfilesThebyteoffsetofthelineThelinecontentsKeyValueInputFormatParseslinesintokey,valpairsEverythinguptothefirsttabcharacterTheremainderofthelineSequenceFileInputFormatAHadoop-specifichigh-performancebinaryformatuser-defineduser-defined各种InputFormatTextInputFormat，默认的格式，每一行是一个单独的记录，并且作为value，文件的偏移值作为keyKeyValueInputFormat，这个格式每一行也是一个单独的记录，但是Key和Value用Tab隔开，是默认的OutputFormat，可以作为中间结果，作为下一步MapReduce的输入。SequenceFileInputFormat基于块进行压缩的格式对于几种类型数据的序列化和反序列化操作用来将数据快速读取到Mapper类中Writable接口Hadoop使用Writable做序列化定义了两个方法二进制写入DataOutput流二进制读取DataInput流Hadoop自带一系列Writable实现，可以满足绝大多数需要可以自定义Writable，控制二进制表示和排序实现Writable接口的例子public

class

MyWritable

implements

Writable

{

//

Some

data

private

int

counter;

private

long

timestamp;

public

void

write(DataOutput

out)

throws

IOException

{

out.writeInt(counter);

out.writeLong(timestamp);

}

public

void

readFields(DataInput

in)

throws

IOException

{

counter

=

in.readInt();

timestamp

=

in.readLong();

}

public

static

MyWritable

read(DataInput

in)

throws

IOException

{

MyWritable

w

=

new

MyWritable();

w.readFields(in);

return

w;

}

}

Writable的Java基本封装Mapper每一个Mapper类的实例生成了一个Java进程（在某一个InputSplit上执行）有两个额外的参数OutputCollector以及Reporter，前者用来收集中间结果，后者用来获得环境参数以及设置当前执行的状态。现在用Mapper.Context提供给每一个Mapper函数，用来提供上面两个对象的功能数据压缩Partition&Shuffle在Map工作完成之后，每一个Map函数会将结果传到对应的Reducer所在的节点，此时，用户可以提供一个Partitioner类，用来决定一个给定的(key,value)对传输的具体位置Combinerconf.setCombinerClass(Reduce.class);是在本地执行的一个Reducer，满足一定的条件才能够执行。Sort传输到每一个节点上的所有的Reduce函数接收到得Key,value对会被Hadoop自动排序（即Map生成的结果传送到某一个节点的时候，会被自动排序）Reduce做用户定义的Reduce操作接收到一个OutputCollector的类作为输出OutputFormat写入到HDFS的所有OutputFormat都继承自FileOutputFormat每一个Reducer都写一个文件到一个共同的输出目录，文件名是part-nnnnn，其中nnnnn是与每一个reducer相关的一个号（partitionid）JobConf.setOutputFormat()RecordWriter用来指导如何输出一个记录到文件中OutputFormatOutputFormat:DescriptionTextOutputFormatDefault;writeslinesin"key\tvalue"formSequenceFileOutputFo