一、MapReduce框架概述1、MapReduce概述关于MapReduce的简单介绍可以参考官方文档的说明Hadoop MapReduce is a software framework for easily writing applications which process vast amounts of data (multi-terabyte>3、MapReduce的核心思想以大数据场景最为典型的WordCount计算为例在计算任务需要统计某个文本中各个单词出现的次数。MapReduce框架的典型计算思想是这样的整体分为MapperShuffle和Reduce计算三个阶段。其中Mapper和Reduce是需要应用自行实现的部分而中间的Shuffle则是由Hadoop来实现不需要进行管理。二、官方示例解析1、官方示例解析下面从一个官方示例来入手。在本地编辑一个简单的文本文件。bash[roothadoop01 ]# vi wdtest.txt my name is roy this is my bigdata lesson welcome to my lesson使用hdfs dfs -put wdtest.txt /input指令将这个文件上传到hdfs的input目录下。执行以下指令启动一个官方提供的MapReduce计算任务。bash[roothadoop01 ~]# hadoop jar /app/hadoop/hadoop-3.2.2/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.2.2.jar wordcount /input /output 2022-01-23 17:22:01,211 INFO client.RMProxy: Connecting to ResourceManager at hadoop02/192.168.65.204:8032 # ---》以下这几行日志就是MapReduce计算的标志性日志 2022-01-23 17:22:03,774 INFO mapreduce.Job: The url to track the job: http://hadoop02:8088/proxy/application_1642929672088_0001/ 2022-01-23 17:22:03,775 INFO mapreduce.Job: Running job: job_1642929672088_0001 ... 2022-01-23 17:22:18,227 INFO mapreduce.Job: map 0% reduce 0% 2022-01-23 17:22:25,431 INFO mapreduce.Job: map 100% reduce 0% 2022-01-23 17:22:32,527 INFO mapreduce.Job: map 100% reduce 100%计算结果会输出到hdfs上的Output目录下。需要启动HDFS和YARN获取源码可以通过反编译任务jar包或从Hadoop源码中获取。3.2.2版本Hadoop源码下载地址https://archive.apache.org/dist/hadoop/common/hadoop-3.2.2/hadoop-3.2.2-src.tar.gz。导入IDEA后直接搜索WordCount即可找到任务的入口执行类。重点关注其main方法javapublic 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 [in...] out); System.exit(2); } Job job Job.getInstance(conf, word count); //寻找执行的jar包 job.setJarByClass(WordCount.class); //Mapper任务类 job.setMapperClass(TokenizerMapper.class); //可选任务对Mapper的数据结果先进行一次本地合并 job.setCombinerClass(IntSumReducer.class); //Reduce任务类 job.setReducerClass(IntSumReducer.class); //设置输出Key类型 job.setOutputKeyClass(Text.class); //设置输出Value类型 job.setOutputValueClass(IntWritable.class); for (int i 0; i otherArgs.length - 1; i) { FileInputFormat.addInputPath(job, new Path(otherArgs[i])); } FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1])); System.exit(job.waitForCompletion(true) ? 0 : 1); }从这个示例中可以了解到MapReduce计算程序的基础开发过程。例如对Job可以设置哪些额外的属性对Mapper任务、Reduce任务有个基本的了解。重点是理解Mapper的输出结果和Reduce的输入结果。注意输入地址和输出地址是需要作为参数配置进去的。例如/input /output。如果在本地执行这两个地址都将是本地文件中的地址。如果要访问HDFS可以配置hdfs://hadoop01:8020/input hdfs://hadoop01:8020/output。2、搭建开发环境这是一个典型的MapReduce任务。这个任务即可以像之前那样打个包提交到Hadoop集群中执行也可以在IDEA直接本地执行。当然本地执行前需要完成Maven编译的整个过程。这个过程很容易出错所以还是建议单独搭建一个Hadoop的MapReduce测试工程。先创建一个Maven项目在pom.xml中引入相关依赖xmldependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.2.2/version /dependency dependency groupIdjunit/groupId artifactIdjunit/artifactId version4.13.2/version scopetest/scope /dependency dependency groupIdorg.slf4j/groupId artifactIdslf4j-log4j12/artifactId version1.7.30/version /dependency然后就可以将WordCount相关的类移入到该测试项目中。这个测试项目将作为我们深入理解MapReduce的演示项目以及调试入口。三、MapReduce计算过程详解1、Mapper任务的执行流程对于Map阶段整体流程如下图阶段一客户端提交任务阶段这一阶段主要做两件事情一是预先对文件进行切片二是提交任务信息到HDFS。文件切片客户端会在提交任务前对待处理的数据(主要是文本)进行预处理文本数据按照BlockSize大小进行拆分。关键源码调试点源码调试从Job的waitForCompletion方法入手。其中submit()方法就是对任务进行提交。文件切割的关键代码在JobSubmitter的writeNewSplits方法。客户端通过InputFormat来计算文件分割。其中isSplitable方法判断文件是否可以拆分。对某些压缩过的文件是不支持拆分的。而getSplits方法实际进行文件拆分。拆分的文件大小默认与HDFS的Block大小一致。默认行为默认会加载TextInputFormat他是FileInputFormat的一个实现类。对于文本都是可以切割的所以直接返回true。然后getSplits方法对文件进行拆分时并不是对文件进行实际切割只是记录一系列InputSplit对象。其中包含的是每个切分块在文件当中的偏移量。这些文件切割信息最终会生成.split文件上传到任务目录中。输入格式InputFormat拆分的结果是(K,V)型的键值对。对于FileInputFormat拆分出的结果中value就是每一行文本而Key就是当前这一行在整个文件中的起始字节偏移量。例如在MapReduce的执行日志中会看到一行这样的日志INFO [org.apache.hadoop.mapred.MapTask] - Processing split:hdfs://hadoop01:8020/input/wdtest.txt:062这里的0和62就是文件切割出来的偏移量。其他InputFormat还有其他的InputFormat的实现类可以通过job.setInputFormatClass();方法进行设置。另一种比较常用的实现类是CombineTextInputFormat。他的切片方式就是只按照文件大小切片不管多少个文件都会是一个单独的切片也就会交给一个MapTask进行后续的处理。切片大小可以通过CombineTextInputFormat.setMaxInputSplitSize方法自行指定。这种切片方式适合于有很多小文件的场景。因为如果小文件很多默认的FileInputFormat实现类就会产生非常多的小分片后续也就启动非常多的MapTask对性能是一种浪费。提交任务信息客户端会将任务信息提交到HDFS当中。这个任务信息会在MapReduce任务执行完成后删除。关键源码调试点在JobSubmitter类的submitJobInternal方法中会获得一个提交任务的地址jobStagingArea指向一个.stage文件。如果是本地执行会指向一个file:开头的本地目录如果是Yarn集群执行会指向一个hdfs:开头的HDFS地址。客户端会负责将MapReduce执行的任务文件上传到HDFS中并通知Runner执行。如果是本地执行会通知LocalJobRunner。Yarn集群则会通知YarnRunner。在这个地方就完成了客户端与Hadoop集群的任务交接。客户端提交的.stage文件夹中主要包含三个文件1job.split文件包含目标文件的切割信息。2job.xml包含当前任务执行的所有配置属性。3wc.jar任务的jar包文件。本地执行时由于所有的类就在当前JVM中所以不需要jar包。通知Runner执行客户端将MapReduce执行的任务文件上传到HDFS后就会通知Runner执行。客户端会通知一个ClientProtocol实现类来执行任务。根据任务执行环境不同会通知不同的实现类。如果是本地执行会通知LocalJobRunner。Yarn集群则会通知YarnRunner。而这两个ClientProtocol实现在接收到任务后会返回给客户端一个集群状态然后在构建任务时就会自动启动任务进行计算。提交完成后将任务状态调整为JobState.RUNNING。到这一步就完成了客户端的整个任务提交过程。后续客户端可以持续跟踪任务的执行进度并打印日志。阶段二MapTask任务执行阶段这一阶段会启动多个MapTask执行Mapper任务。MapTask的数量取决于上一阶段分片的个数。4.读取数据MapTask会先将文件读取成K-V类型的元数据再交由Mapper执行。*.stage文件夹中的job.xml文件实际上是在任务执行阶段由ClientProtocol类在执行阶段生成的。这也比较好理解只有在执行阶段才会去获取所有的执行参数。* Job执行过程中会根据.split文件中的切片个数启动对应的MapTask。5.执行Map方法每一个Block会启动一个MapperTask来处理并最终通过OutputCollector将分散的处理结果收集起来。* MapperTask执行map方法时的(K,V)参数都是由RecordReader对象根据之前的.split文件读取到的内容。RecordReader是一个抽象类具体的实现子类是由之前的InputFormat实现类的createRecordReader方法产生。默认的TextInputFormat就会产生一个LineRecordReader这个实现类会一行一行的读取文件。而CombineTextInputFormat对象会产生一个CombineFileRecordReader。6.后续计算Mapper计算过程中会定时的收集任务完成率反馈给客户端。2、Shuffle阶段的执行流程在Map方法执行之后到Reduce方法执行之前还有一段数据处理的过程。这个过程称为Shuffle。这是整个MapReduce最为核心的一段过程。整个Shuffle的过程如下分区 (Partition)Mapper阶段输出的数据会经过Partition进行分区给每个数据标记上他的分区信息。作用将数据按照条件输出到不同的文件(分区)当中。这些数据将进入Reduce中处理所以每个Reduce接收到的数据都是key相同的。默认实现分区操作是通过Partitioner接口来实现的。默认加载的是HashPartitioner这个实现类会按key的hashcode分组相同的key会分到一个组。但是用户没法控制某一个key具体会分到哪个分区。自定义分区用户可以通过实现Partitioner的方式自定义分区实现然后通过job.setPartitionerClass方法指定。自定义分区时要注意生成的分区号必须从0开始逐一增加不能有跳过。环形缓冲区与溢写分区后的数据会收集进入到环形缓冲区进行缓存。数据收集到一定数量后会从缓冲区溢写到磁盘当中。环形缓冲区是Hadoop中的一块核心内存由MapOutputBuffer进行管理。默认大小是100M。他的作用就是对Mapper产生的数据进行缓存缓存到一定数量后按照分区将数据写到磁盘中。环形缓冲区有一个默认的容量阈值capacity默认是80%。工作机制环形缓冲区在收集数据时会从内存块的头部位置开始同时记录数据的序列化结果(kvbuffer中)和数据的索引信息(kvmeta中)。当环形缓冲区中的数据容量达到他的容量阈值(80%)后环形缓冲区就会改为从内存块的尾部位置开始反向记录数据。而原有的前80%内存块中的数据就会开始往磁盘中进行溢写并逐步开始释放内存。同样当从尾部开始反写的数据容量超过80%后又会再次反过来从内存头部开始写数据然后重复之前的过程。内存块循环使用这也正是环形缓冲区这个名称的来源。如果数据反向写入的速度比内存溢写的速度快那么这时数据写入操作就会暂停等待释放内存。排序溢写出来的数据是按照分区整合到一起的。这时会将各个分区数据进行排序。从环形缓冲区溢写出来的数据是按照分区进行了划分的。相同分区的数据会写入到同一个文件当中。刚从环形缓冲区溢写出来的数据是按分区进行了分组但是组内并不是有序的。这时会使用快排算法对分区内的数据进行排序。合并 (Combine) 与归并Combine操作是Shuffle过程中的一个可选操作由job.setCombinerClass方法指定。在环形缓冲区MapOutputBuffer中也会记录Combiner。Combine是对已经局部有序的多个溢写文件按照key再次进行整合排序这个过程中使用的是归并排序的算法。归并归并后的数据文件可以选择进行第二次Combine操作进行局部合并。形成最终的局部有序的分区数据集合。Combine的作用只是让数据在进入Reduce处理之前先进行一下局部数据合并。这样做并不会影响最终的计算结果。其最大的作用在于减少网络传输的数据量提升计算的性能。压缩分区数据集合可以选择进行压缩最终保存到磁盘当中完成整个Shuffle过程。作用减少网络传输以及减少存盘文件大小提升IO。压缩过程主要有两个执行点一个是Mapper任务输出另一个是Reduce任务输出。配置这两个阶段默认都是不启用压缩的如果要启用数据压缩需要在mapred-site.xml中配置参数例如mapreduce.map.output.compress、mapreduce.map.output.compress.codec等。压缩算法Hadoop中提供了多种数据压缩的算法他们都是CompressionCodec接口的实现类。GZip压缩比率最高但是不支持Split压缩解压的速度一般。BZip2压缩比率也挺高支持Split压缩解压速度最慢。Snappy压缩比率一般不支持Split压缩解压的速度最快。自定义企业也可以按照Hadoop的接口要求自行定制数据压缩算法。不过通常来说数据压缩都会带来额外的计算开销所以如果是计算密集型的任务就不太建议增加数据压缩。让计算资源更多的投入到业务计算中。数据落盘Shuffle过程完成后数据只负责落盘到磁盘文件当中。后续由Reduce任务主动来拉取数据。3、Reduce阶段的执行流程Reduce阶段主要是会对之前Mapper阶段的任务进行聚合。对于一个MapReduce计算任务来说其实Reduce也是一个可选的任务阶段。整体步骤如下启动时机Reduce通常在Mapper任务执行完成后再启动。在MapReduce的执行日志中能够跟踪到Mapper任务和Reduce任务的执行进度。通常情况下Reduce需要收集Mapper输出完的结果所以需要等所有的Mapper任务执行完成后再启动。但是Mapreduce框架也会根据计算情况进行优化。在某些Reduce任务所依赖的Mapper任务全部完成了之后这些Reduce可以提前启动而不用等待所有的Mapper任务执行完成。拉取数据Reduce主动去磁盘中拉取相关的数据。MapTask的并行度是由切片个数决定的而ReduceTask的并行度则不同可以通过setNumReduceTasks方法手动指定默认是1个。MapTask将数据输出到磁盘ReduceTask主动去磁盘上远程拷贝一片数据。如果数据大小超过了一定阈值就会将一部分数据写入到磁盘当中。排序之后Reduce会对数据进行分组将相同key的数据整合到一起。之后才进入用户自定义的Reduce方法中进行聚合。合并数据Reduce对获取到的数据进行合并。ReduceTask在获取到数据之后会对内存和磁盘中的数据进行合并。将相同key的数据整合在一起而Value整合成一个数组。而在合并过程中为了将相同的key快速合并到一起ReduceTask还会对数据再进行一次排序排序之后的数据针对相同的key局部有序。执行Reduce方法与输出进入用户自定义的reduce()方法。每个Reduce任务各自对数据进行计算并将收集到的结果输出到最终文件中。每个Reduce接收到的数据是按key局部有序的但是多个Reduce任务之间的key并不能保证是有序的。每个Reduce都将结果输出到自己对应的目标文件中输出的结果同样是按照key在每个文件内局部有序整体并不保证有序。Reduce的计算结果最终交由FileOutputFormat的实现类来输出。默认加载的是TextOutputFormat。他会产生一个LineRecordWriter将每一个KV对向目标文件输出一行。用户同样可以使用job.setOutputFormatClass()方法选择其他的结果输出器也可以定制自定义的结果输出格式。4、MapReduce流程整体总结这样我们将MapReduce任务的各个阶段详细进行拆解后就形成了MapReduce任务更加详细的整体细节流程图接下来我们重点梳理一下有哪些过程是可以由用户参与的。输入数据接口InputFormat默认使用的实现类是TextInputFormatTextInputFormat的功能逻辑是一次读一行文本然后将该行的起始偏移量作为key行内容作为value返回。CombineTextInputFormat可以把多个小文件合并成一个切片处理提高处理效率。逻辑处理接口Mapper用户根据业务需求实现其中三个方法map()setup()cleanup()Partitioner分区默认的实现方式是HashPartitioner逻辑是根据key的哈希值和numReduces来返回一个分区号key.hashCode() Integer.MAXVALUE % numReduces。数据分配均匀但是由于HashCode是很难预估的所以预先无法知道具体的分区结果。如果业务上有特别的需求可以自定义分区。Comparable排序当我们用自定义的对象作为key来输出时就必须要实现WritableComparable接口重写其中的compareTo()方法。Combiner合并Combiner合并可以提高程序执行效率减少IO传输。但是使用时必须不能影响原有的业务处理结果。数据压缩算法CompressionCodec用户可以定义数据压缩算法但是需要注意压缩算法是否支持数据分片。逻辑处理接口Reducer用户根据业务需求实现其中三个方法reduce()setup()cleanup()输出数据接口OutputFormat默认实现类是TextOutputFormat功能逻辑是将每一个KV对向目标文本文件输出一行。用户还可以自定义OutputFormat。四、深入理解1、数据倾斜问题数据倾斜是大数据计算中非常常见的一个问题。其原因就是由于数据分布不均匀可能造成某些partition的数据非常多而某些partition的数据非常少。分配给不同的ReduceTask处理后数据多的ReduceTask计算任务就非常繁忙而数据少的ReduceTask计算任务就非常简单。这对于集群资源其实是一种浪费。体现出来的效果就是Reduce阶段的进展非常不平衡前期进度比较快后期进度非常慢。解决数据倾斜的问题核心思想还是要调整数据的分区逻辑让各个分区的数据尽量平均一点。调整的方式主要有两种调整Reduce的数量这样可以一定程度上分担ReduceTask的计算压力。并且默认的HashPartitioner实现在调整了Reduce的数量后也会对数据重新进行分区。但是这种方法的坏处是HashPartitioner根据key的hashCode和ReduceTask来计算分区由于hashCode是很难把握的所以并不能保证调整后的分区就一定能够更加平均。可以在简单的WordCount计算过程中调整numReduceTask属性观察结果job.setNumReduceTasks(2);自定义分区给MapReduce计算设置一个自定义的Partitioner实现让数据按照自定义的逻辑进行分区。通过自定义分区应用程序可以根据数据的具体情况灵活调整分区逻辑保证数据分区更加平均。另外自定义分区最终会影响到ReduceTask的输出文件所以也可以让MapReduce的输出文件更有意义。例如可以按照英文单词的首个字母分组。首字母相同的单词就会固定输出到同一个文件当中。2、自定义对象序列化示例中每个计算步骤比如MapTask和ReduceTask他们的输入和输出类型是一些基础类型的数据如果想要传输一些自定义的数据类型比如自己的POJO对象那就需要了解自定义序列化。这也是开发中非常常见的功能。MapReduce对计算过程中的Key和Value都需要通过泛型进行严格的定义。默认对各种基础数据类型都提供了对应的封装类型如IntWritable,LongWritable,FloatWritable,DoubleWritable,BooleanWritable,ByteWritable,Text等。其中String比较特殊一点对应的封装类是Text。可以看到MapReduce对于基础数据类型都提供了对应的Writable封装类型。这些封装类型都是实现自WritableComparable接口。javapublic interface WritableComparableT extends Writable, ComparableT { }这个接口实现了两个接口Hadoop定义的Writable接口以及JDK提供的Comparable接口。其中Comparable接口就是给数据提供比较的功能。这是因为MapReduce计算中需要多次对数据进行排序而排序就需要有Comparable接口支持。Writable接口则是Hadoop定义的一个序列化实现接口。这个接口的作用就是将数据转换成二进制的数据流这些数据流才可以在网络上进行传输。javapublic interface Writable { void write(DataOutput out) throws IOException; void readFields(DataInput in) throws IOException; }Writable提供了两个方法write方法定义如何将数据对象写成二进制数据流readFields定义如何将二进制数据流读取成数据对象。如果我们需要计算自定义的数据对象那么数据对象就需要自行实现Writable接口的这两个方法。在Writable接口的源码注释当中就提供了一个简单的示例javapublic class MyWritable implements Writable { // Some data private int counter; private long timestamp; // Default constructor to allow (de)serialization MyWritable() { } 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; } }3、自定义输出文件MapReduce的输出文件都收Part-xxx这样的格式他是按照Reduce拆分输出的结果。这样的输出结果通常没有太多的意义。如果想要自定义输出文件也可以通过自定义OutputFormat的方式进行自定义。4、Hadoop中的RPC框架如果想要深入Hadoop的源码从MapReduce计算任务为入口其实是一个比较好的方法。整个Hadoop的源码相当庞大并且有很多的内部任务所以通常很难找到调试代码的主线。而从MapReduce计算任务的源码开始调试就比较容易找到主线。在跟踪Hadoop源码的时候他的RPC框架需要提前了解下要不然Hadoop集群内部的大量RPC请求会非常难以跟踪。而Hadoop基于Netty实现了一套非常简洁小巧的RPC框架。这个小框架脱离Hadoop框架也是很有借鉴意义的。使用Hadoop的RPC框架只需要引入Hadoop的common模块即可。引入如下hadoop-client的maven依赖就会包含common模块。xmldependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.2.2/version /dependency然后Hadoop的RPC框架整体由三个部分构成服务端客户端协议。Hadoop中的协议层是一个非常简单灵活的层面只需要定义一个带版本的接口以及服务实现类即可。javapublic interface ClientNamenodeProtocol { public long versionID 1L; public String getMetaData(String path); } //模拟NameNode根据文件路径返回文件所在的Block位置 public class MyNamenode implements ClientNamenodeProtocol { Override public String getMetaData(String path) { return path : 3 - {BLK_1,BLK_2} ......; } }Hadoop中的协议是很自由的接口。内部提供了一个VersionedProtocol但是并不强制要求继承。在服务端就可以根据这个协议接口和实现类直接创建一个服务端。javapublic class RPCServer { public static void main(String[] args) throws IOException { final RPC.Builder builder new RPC.Builder(new Configuration()); builder.setBindAddress(localhost); builder.setPort(8888); builder.setProtocol(ClientNamenodeProtocol.class) .setInstance(new MyNamenode()); final RPC.Server server builder.build(); server.start(); } }启动后就会绑定本机的8888端口等待客户端响应。而客户端就可以直接请求协议中对应的方法调用服务端的实现类。javapublic class RPCClient { public static void main(String[] args) throws IOException { final ClientNamenodeProtocol proxy RPC.getProxy(ClientNamenodeProtocol.class, 1L, new InetSocketAddress(localhost, 8888), new Configuration()); final String metaData proxy.getMetaData(/myNameNode); System.out.println(metaData); } }这个RPC框架在Hadoop底层应用非常广泛例如在NameNode上就会启动一个RPC服务端来接收DataNode的心跳请求。在NameNodeRpcServer.java源码中在他的构造方法中就有这样一段源码javaserviceRpcServer new RPC.Builder(conf) .setProtocol(org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolPB.class) .setInstance(clientNNPbService) .setBindAddress(bindHost) .setPort(serviceRpcAddr.getPort()) .setNumHandlers(serviceHandlerCount) .setVerbose(false) .setSecretManager(namesystem.getDelegationTokenSecretManager()) .build();5、Hadoop StreamingHadoop Streaming是Hadoop新推出的一个工具集。这个工具集并不是提供流式计算的功能而是允许以命令行方式代替千篇一律的Driver代码。通过Streaming工具可以简化MapReduce任务的部署。其实从之前的编码过程可以看到对于一个完整的MapReduce计算任务重要的业务逻辑都是由Mapper、Reducer、Partitioner、OutputFormat等这些组件来提供的而启动任务的Driver其实只是相当于一个组件粘合剂并没有实际的业务逻辑。Hadoop Streaming就允许以命令行的方式来启动一个Driver。这样可以减少提交任务的Jar包大小因为Driver类就不需要了。并且通过Hadoop Streaming工具可以扩充各个功能组件的具体实现。Linux的指令工具以及Python语言开发的组件也都可以自由组合到MapReduce计算任务当中。不过企业一般都是习惯于用一个完整jar包的方式来提交任务所以这个工具其实用得并不是很多。这里就只做简单的介绍。5.1 零开发完成一个WordCount计算。直接执行以下指令就可以启动一个MapReduce任务。bash[roothadoop01 ~]# mapred streaming \ -input /input2 \ -output /output2 \ -mapper /bin/cat \ -reducer /usr/bin/wc这个指令就会启动一个MapReduce计算统计HDFS上/input2目录下的文件输出到/output2目录。其中mapper直接使用Linux的cat指令完成而reduce直接使用Linux的wc指令完成所以统计完成的输出结果是这样的36 620 4228。这表示/input2下面的一个文件行数为36、单词数620、字节数4228。5.2 Hadoop Streaming指令介绍Hadoop Streaming的基础指令格式是这样的bashmapred streaming [genericOptions] [streamingOptions]其中genericOptions表示一些任务相关的配置信息。主要有以下几个配置ParameterOptional/RequiredDescription-conf configuration_fileOptionalSpecify an application configuration file-D propertyvalueOptionalUse value for given property-fs host:port or localOptionalSpecify a namenode-filesOptionalSpecify comma-separated files to be copied to the Map/Reduce cluster-libjarsOptionalSpecify comma-separated jar files to include in the classpath-archivesOptionalSpecify comma-separated archives to be unarchived on the compute machines例如-D mapreduce.job.reduces 0指定reduceTask个数。-files允许访问HDFS上的文件。-archives可以引入HDFS上的一个任务jar包就不用提交到Hadoop了。streamingOptions则是一组具体的执行指令。streamingOptions需要在genericOptions之后指定。可选的指令如下ParameterOptional/RequiredDescription-input directoryname or filenameRequiredInput location for mapper-output directorynameRequiredOutput location for reducer-mapper executable or JavaClassNameOptionalMapper executable. If not specified, IdentityMapper is used as the default-reducer executable or JavaClassNameOptionalReducer executable. If not specified, IdentityReducer is used as the default... (其他参数详见文档表格)其实对照Driver的实现过程很多的配置都一目了然。像combinerpartitionermapperreducer等。与Driver的不同之处在于这些功能组件都可以指定shell指令或者python脚本。例如以下指令可以引入一个Parititoner。而这个Partition类所在的jar包即可以上传到Hadoop的安装目录下也可以上传到HDFS中通过-archives引入。bashmapred streaming \ -D stream.map.output.field.separator. \ -D stream.num.map.output.key.fields4 \ -D map.output.key.field.separator. \ -D mapreduce.partition.keypartitioner.options-k1,2 \ -D mapreduce.job.reduces12 \ -input /input \ -output /output3 \ -mapper /bin/cat \ -reducer /bin/cat \ -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner