Mapreduce基本程序

1.Mapper里面的map方法

public void map(Object key,Text value,Context context) throws IOException,InterruptedException{...}

map()方法里面有三个参数，Object key,Text value就是输入的key和value,第三个参数Context context可以记录输入的key和value，此外context还会记录map运算的状态。

2.Reducer里面的reduce()方法

pubilic void reduce(Text key,Iterable<IntWritable>values,Context context) throws IOException,InterruptedException{...}

reduce()函数的输入也是一个key/value的形式，不过它的value是一个迭代器的形式Iterable<IntWritable>values，也就是说reduce中的values是一个key对应一组的值得value。

3.main()函数

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

Configuration conf = new Configuration()

String[] otherArgs =

new GenericOptionsParser(conf, args).getRemainingArgs()

if (otherArgs.length != 2) {

System.err.println("Usage: wordcount <in><out>")

System.exit(2)

}

Job job = new Job(conf, "word count")//构建一个job

job.setJarByClass(WordCount.class)

job.setMapperClass(TokenizerMapper.class)

job.setCombinerClass(IntSumReducer.class)

job.setReducerClass(IntSumReducer.class)

job.setOutputKeyClass(Text.class)

job.setOutputValueClass(IntWritable.class)

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

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

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

}

（1）Configuration conf=new Configuration()//初始化

运行mapreduce程序前都要初始化Configuration，该类主要是读取mapreduce系统配置信息，这些信息包括hdfs还有mapreduce，也就是安装hadoop时候的配置文件例如：core-site.xml、hdfs-site.xml和mapred-site.xml等等文件里的信息。

程序员开发mapreduce时候只是在填空，在map函数和reduce函数里编写实际进行的业务逻辑，其它的工作都是交给mapreduce框架自己 *** 作的，但是至少我们要告诉它怎么 *** 作啊，比如hdfs在哪里啊，mapreduce的jobstracker在哪里啊，而这些信息就在conf包下的配置文件里。

（2） Job job = new Job(conf, "word count")

在mapreduce框架里一个mapreduce任务也叫mapreduce作业也叫做一个mapreduce的job。

具体的map和reduce运算就是task了，这里我们构建一个job，构建时候有两个参数，一个是conf，一个是这个job的名称。

（3） job.setOutputKeyClass(Text.class)

job.setOutputValueClass(IntWritable.class)

定义输出的key/value的类型，也就是最终存储在hdfs上结果文件的key/value的类型。

（4）

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

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

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

第一行就是构建输入的数据文件

第二行是构建输出的数据文件

最后一行是如果job运行成功了，程序就正常退出。

FileInputFormat和FileOutputFormat可以设置输入输出文件路径，

mapreduce计算时候:

输入文件必须存在，不然mapreduce任务直接退出。

输出一般是一个文件夹，而且该文件夹不能存在。

用户配置并将一个Hadoop作业提到Hadoop框架中，Hadoop框架会把这个作业分解成一系列map tasks 和reduce tasks。Hadoop框架负责task分发和执行，结果收集和作业进度监控。

下图给出了一个作业从开始执行到结束所经历的阶段和每个阶段被谁控制（用户 or Hadoop框架）。

下图详细给出了用户编写MapRedue作业时需要进行那些工作以及Hadoop框架自动完成的工作：

在编写MapReduce程序时，用户分别通过InputFormat和OutputFormat指定输入和输出格式，并定义Mapper和Reducer指定map阶段和reduce阶段的要做的工作。在Mapper或者Reducer中，用户只需指定一对key/value的处理逻辑，Hadoop框架会自动顺序迭代解析所有key/value，并将每对key/value交给Mapper或者Reducer处理。表面上看来，Hadoop限定数据格式必须为key/value形式，过于简单，很难解决复杂问题，实际上，可以通过组合的方法使key或者value（比如在key或者value中保存多个字段，每个字段用分隔符分开，或者value是个序列化后的对象，在Mapper中使用时，将其反序列化等）保存多重信息，以解决输入格式较复杂的应用。

2.2 用户的工作

用户编写MapReduce需要实现的类或者方法有：

（1） InputFormat接口

用户需要实现该接口以指定输入文件的内容格式。该接口有两个方法

其中getSplits函数将所有输入数据分成numSplits个split，每个split交给一个map task处理。getRecordReader函数提供一个用户解析split的迭代器对象，它将split中的每个record解析成key/value对。

Hadoop本身提供了一些InputFormat：

（2）Mapper接口

用户需继承Mapper接口实现自己的Mapper，Mapper中必须实现的函数是

void map(K1 key,

V1 value,

OutputCollector<K2,V2>output,

Reporter reporter

) throws IOException

其中，<K1 V1>是通过Inputformat中的RecordReader对象解析处理 的，OutputCollector获取map()的输出结果，Reporter保存了当前task处理进度。

Hadoop本身提供了一些Mapper供用户使用：

（3）Partitioner接口

用户需继承该接口实现自己的Partitioner以指定map task产生的key/value对交给哪个reduce task处理，好的Partitioner能让每个reduce task处理的数据相近，从而达到负载均衡。Partitioner中需实现的函数是

getPartition(  K2   key, V2 value, int numPartitions)

该函数返回<K2 V2>对应的reduce task ID。

用户如果不提供Partitioner，Hadoop会使用默认的（实际上是个hash函数）。

（4）Combiner

Combiner使得map task与reduce task之间的数据传输量大大减小，可明显提高性能。大多数情况下，Combiner与Reducer相同。

（5）Reducer接口

用户需继承Reducer接口实现自己的Reducer，Reducer中必须实现的函数是

Hadoop本身提供了一些Reducer供用户使用：

（6）OutputFormat

用户通过OutputFormat指定输出文件的内容格式，不过它没有split。每个reduce task将其数据写入自己的文件，文件名为part-nnnnn，其中nnnnn为reduce task的ID。

Hadoop本身提供了几个OutputFormat:

3. 分布式缓存

Haoop中自带了一个分布式缓存，即DistributedCache对象，方便map task之间或者reduce task之间共享一些信息，比如某些实际应用中，所有map task要读取同一个配置文件或者字典，则可将该配置文件或者字典放到分布式缓存中。

4. 多语言编写MapReduce作业

Hadoop采用java编写，因而Hadoop天生支持java语言编写作业，但在实际应用中，有时候，因要用到非java的第三方库或者其他原因，要采用C/C++或者其他语言编写MapReduce作业，这时候可能要用到Hadoop提供的一些工具。

一、 首先要知道此前提 转载

若在windows的Eclipse工程中直接启动mapreduc程序，需要先把hadoop集群的配置目录下的xml都拷贝到src目录下，让程序自动读取集群的地址后去进行分布式运行(您也可以自己写java代码去设置job的configuration属性)。

若不拷贝，工程中bin目录没有完整的xml配置文件，则windows执行的mapreduce程序全部通过本机的jvm执行，作业名也是带有“local"字眼的作业，如 job_local2062122004_0001。 这不是真正的分布式运行mapreduce程序。

估计得研究org.apache.hadoop.conf.Configuration的源码，反正xml配置文件会影响执行mapreduce使用的文件系统是本机的windows文件系统还是远程的hdfs系统还有影响执行mapreduce的mapper和reducer的是本机的jvm还是集群里面机器的jvm

二、 本文的结论

第一点就是： windows上执行mapreduce，必须打jar包到所有slave节点才能正确分布式运行mapreduce程序。（我有个需求是要windows上触发一个mapreduce分布式运行）

第二点就是： Linux上，只需拷贝jar文件到集群master上,执行命令hadoop jarPackage.jar MainClassName即可分布式运行mapreduce程序。

第三点就是： 推荐使用附一，实现了自动打jar包并上传，分布式执行的mapreduce程序。

附一、 推荐使用此方法：实现了自动打jar包并上传，分布式执行的mapreduce程序：

请先参考博文五篇：

Hadoop作业提交分析（一）~~（五）

引用博文的附件中EJob.java到你的工程中，然后main中添加如下方法和代码。

public static File createPack() throws IOException {

File jarFile = EJob.createTempJar("bin")

ClassLoader classLoader = EJob.getClassLoader()

Thread.currentThread().setContextClassLoader(classLoader)

return jarFile

}

在作业启动代码中使用打包：

Job job = Job.getInstance(conf, "testAnaAction")

添加：

String jarPath = createPack().getPath()

job.setJar(jarPath)

即可实现直接run as java application 在windows跑分布式的mapreduce程序，不用手工上传jar文件。

附二、得出结论的测试过程

（未有空看书，只能通过愚笨的测试方法得出结论了）

一. 直接通过windows上Eclipse右击main程序的java文件，然后"run as application"或选择hadoop插件"run on hadoop"来触发执行MapReduce程序的测试。

1，如果不打jar包到进集群任意linux机器上，它报错如下：

[work] 2012-06-25 15:42:47,360 - org.apache.hadoop.mapreduce.Job -10244 [main] INFO org.apache.hadoop.mapreduce.Job - map 0% reduce 0%

[work] 2012-06-25 15:42:52,223 - org.apache.hadoop.mapreduce.Job -15107 [main] INFO org.apache.hadoop.mapreduce.Job - Task Id : attempt_1403517983686_0056_m_000000_0, Status : FAILED

Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found

at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1720)

at org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:186)

at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:721)

at org.apache.hadoop.mapred.MapTask.run(MapTask.java:339)

at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:162)

at java.security.AccessController.doPrivileged(Native Method)

at javax.security.auth.Subject.doAs(Subject.java:415)

at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1491)

at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:157)

Caused by: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found

at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:1626)

at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1718)

... 8 more

# Error:后重复三次

2012-06-25 15:44:53,234 - org.apache.hadoop.mapreduce.Job -37813 [main] INFO org.apache.hadoop.mapreduce.Job - map 100% reduce 100%

现象就是：报错，无进度，无运行结果。

2，拷贝jar包到“只是”集群master的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse "run as application"和通过hadoop插件"run on hadoop"来触发执行，它报错同上。

现象就是：报错，无进度，无运行结果。

3，拷贝jar包到集群某些slave的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse "run as application"和通过hadoop插件"run on hadoop"来触发执行

和报错：

Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found

at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1720)

at org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:186)

和报错：

Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountReducer not found

现象就是：有报错，但仍然有进度，有运行结果。

4，拷贝jar包到集群所有slave的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse "run as application"和通过hadoop插件"run on hadoop"来触发执行：

现象就是：无报错，有进度，有运行结果。

第一点结论就是： windows上执行mapreduce，必须打jar包到所有slave节点才能正确分布式运行mapreduce程序。

二 在Linux上的通过以下命令触发MapReduce程序的测试。

hadoop jar $HADOOP_HOME/share/hadoop/mapreduce/bookCount.jar bookCount.BookCount

1，只拷贝到master，在master上执行。

现象就是：无报错，有进度，有运行结果。

2，拷贝随便一个slave节点,在slave上执行。

现象就是：无报错，有进度，有运行结果。

但某些节点上运行会报错如下，且运行结果。：

14/06/25 16:44:02 INFO mapreduce.JobSubmitter: Cleaning up the staging area /tmp/hadoop-yarn/staging/hduser/.staging/job_1403517983686_0071

Exception in thread "main" java.lang.NoSuchFieldError: DEFAULT_MAPREDUCE_APPLICATION_CLASSPATH

at org.apache.hadoop.mapreduce.v2.util.MRApps.setMRFrameworkClasspath(MRApps.java:157)

at org.apache.hadoop.mapreduce.v2.util.MRApps.setClasspath(MRApps.java:198)

at org.apache.hadoop.mapred.YARNRunner.createApplicationSubmissionContext(YARNRunner.java:443)

at org.apache.hadoop.mapred.YARNRunner.submitJob(YARNRunner.java:283)

at org.apache.hadoop.mapreduce.JobSubmitter.submitJobInternal(JobSubmitter.java:415)

at org.apache.hadoop.mapreduce.Job$10.run(Job.java:1268)

at org.apache.hadoop.mapreduce.Job$10.run(Job.java:1265)

at java.security.AccessController.doPrivileged(Native Method)

at javax.security.auth.Subject.doAs(Subject.java:415)

at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1491)

at org.apache.hadoop.mapreduce.Job.submit(Job.java:1265)

at org.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:1286)

at com.etrans.anaSpeed.AnaActionMr.run(AnaActionMr.java:207)

at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70)

at com.etrans.anaSpeed.AnaActionMr.main(AnaActionMr.java:44)

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

at java.lang.reflect.Method.invoke(Method.java:606)

at org.apache.hadoop.util.RunJar.main(RunJar.java:212)

第二点结论就是： Linux上，只需拷贝jar文件到集群master上,执行命令hadoop jarPackage.jar MainClassName即可分布式运行mapreduce程序。

---