MapReduce思想MapReduce思想在生活中处处可见。我们或多或少都曾接触过这种思想。MapReduce的思想核心是分而治之，充分利用了并行处理的优势。即使是发布过论文实现分布式计算的谷歌也只是实现了这种思想，而不是自己原创。MapReduce任务过程是分为两个处理阶段：Map阶段：Map阶段的主要作用是“分”，即把复杂的任务分解为若干个“简单的任务”来并行处理。Map阶段的这些任务可以并行计算，彼此间没有依赖关系。Reduce阶段：Reduce阶段的主要作用是“合”，即对map阶段的结果进行全局汇总。再次理解MapReduce的思想官方WordCount案例源码解析经过查看分析官方WordCount案例源码我们发现一个统计单词数量的MapReduce程序的代码由三个部分组成：Mapper类Reducer类运行作业的代码（Driver）Mapper类继承了org.apache.hadoop.mapreduce.Mapper类重写了其中的map方法，Reducer类继承了org.apache.hadoop.mapreduce.Reducer类重写了其中的reduce方法。重写的Map方法作用：map方法其中的逻辑就是用户希望mr程序map阶段如何处理的逻辑；重写的Reduce方法作用：reduce方法其中的逻辑是用户希望mr程序reduce阶段如何处理的逻辑；Hadoop序列化为什么进行序列化？序列化主要是我们通过网络通信传输数据时或者把对象持久化到文件，需要把对象序列化成二进制的结构。观察源码时发现自定义Mapper类与自定义Reducer类都有泛型类型约束，比如自定义Mapper有四个形参类型，但是形参类型并不是常见的java基本类型。为什么Hadoop要选择建立自己的序列化格式而不使用java自带serializable？序列化在分布式程序中非常重要，在Hadoop中，集群中多个节点的进程间的通信是通过RPC（远程过程调用：Remote Procedure Call）实现；RPC将消息序列化成二进制流发送到远程节点，远程节点再将接收到的二进制数据反序列化为原始的消息，因此RPC往往追求如下特点：紧凑:数据更紧凑，能充分利用网络带宽资源快速:序列化和反序列化的性能开销更低Hadoop使用的是自己的序列化格式Writable,它比java的序列化serialization更紧凑速度更快。一个对象使用Serializable序列化后，会携带很多额外信息比如校验信息，Header，继承体系等。Java基本类型与Hadoop常用序列化类型Java基本类型Hadoop Writable类型booleanBooleanWritablebyteByteWritableintIntWritablefloatFloatWritablelongLongWritabledoubleDoubleWritableStringTextmapMapWritablearrayArrayWritableMapReduce编程规范及示例编写Mapper类用户自定义一个Mapper类继承Hadoop的Mapper类Mapper的输入数据是KV对的形式（类型可以自定义）Map阶段的业务逻辑定义在map()方法中Mapper的输出数据是KV对的形式（类型可以自定义）注意：map()方法是对输入的一个KV对调用一次！！Reducer类用户自定义Reducer类要继承Hadoop的Reducer类Reducer的输入数据类型对应Mapper的输出数据类型（KV对）Reducer的业务逻辑写在reduce()方法中Reduce()方法是对相同K的一组KV对调用执行一次Driver阶段创建提交YARN集群运行的Job对象，其中封装了MapReduce程序运行所需要的相关参数入输入数据路径，输出数据路径等，也相当于是一个YARN集群的客户端，主要作用就是提交我们MapReduce程序运行。WordCount代码实现需求在给定的文本文件中统计输出每一个单词出现的总次数输入数据：wc.txt;输出：apache 2clickhouse 2hadoop 1mapreduce 1spark 2xiaoming 1具体步骤按照MapReduce编程规范，分别编写Mapper，Reducer，Driver。新建maven工程导入hadoop依赖<dependencies> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.8.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.9.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.9.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.9.2</version> </dependency> </dependencies> <build> <plugins> <plugin> <artifactId>maven-compiler-plugin</artifactId> <version>2.3.2</version> <configuration> 1.8 <target>1.8</target> </configuration> </plugin> <plugin> <artifactId>maven-assembly-plugin </artifactId> <configuration> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> </plugins> </build>添加log4j.propertieslog4j.rootLogger=INFO, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n log4j.appender.logfile=org.apache.log4j.FileAppender log4j.appender.logfile.File=target/spring.log log4j.appender.logfile.layout=org.apache.log4j.PatternLayout log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n整体思路梳理（仿照源码）Map阶段：map()方法中把传入的数据转为String类型根据空格切分出单词输出<单词，1>Reduce阶段：汇总各个key(单词)的个数，遍历value数据进行累加输出key的总数Driver获取配置文件对象，获取job对象实例指定程序jar的本地路径指定Mapper/Reducer类指定Mapper输出的kv数据类型指定最终输出的kv数据类型指定job处理的原始数据路径指定job输出结果路径提交作业编写Mapper类import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{ Text k = new Text(); IntWritable v = new IntWritable(1); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 1 获取一行 String line = value.toString(); // 2 切割 String[] words = line.split(" "); // 3 输出 for (String word : words) { k.set(word); context.write(k, v); } } }继承的Mapper类型选择新版本API：编写Reducer类import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{ int sum; IntWritable v = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException { // 1 累加求和 sum = 0; for (IntWritable count : values) { sum += count.get(); } // 2 输出 v.set(sum); context.write(key,v); } }选择继承的Reducer类：编写Driver驱动类import java.io.IOException; 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; public class WordcountDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { // 1 获取配置信息以及封装任务 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 2 设置jar加载路径 job.setJarByClass(WordcountDriver.class); // 3 设置map和reduce类 job.setMapperClass(WordcountMapper.class); job.setReducerClass(WordcountReducer.class); // 4 设置map输出 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 5 设置最终输出kv类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 6 设置输入和输出路径 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 7 提交 boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }运行任务本地模式直接Idea中运行驱动类即可idea运行需要传入参数：在program arguments设置参数运行结束，去到输出结果路径查看结果注意本地idea运行mr任务与集群没有任何关系，没有提交任务到yarn集群，是在本地使用多线程方式模拟的mr的运行。Yarn集群模式1）把程序打成jar包，改名为wc.jar;上传到Hadoop集群选择合适的Jar包准备原始数据文件，上传到HDFS的路径，不能是本地路径，因为跨节点运行无法获取数据！！2）启动Hadoop集群（Hdfs,Yarn）3）使用Hadoop 命令提交任务运行hadoop jar wc.jar com.test.wordcount.WordcountDriver /user/test/input /user/test/outputYarn集群任务运行成功展示图序列化Writable接口基本序列化类型往往不能满足所有需求，比如在Hadoop框架内部传递一个自定义bean对象，那么该对象就需要实现Writable序列化接口。实现Writable序列化步骤如下必须实现Writable接口反序列化时，需要反射调用空参构造函数，所以必须有空参构造public CustomBean() { super(); }重写序列化方法@Override public void write(DataOutput out) throws IOException { .... }重写反序列化方法@Override public void readFields(DataInput in) throws IOException { .... }反序列化的字段顺序和序列化字段的顺序必须完全一致方便展示结果数据，需要重写bean对象的toString()方法，可以自定义分隔符如果自定义Bean对象需要放在Mapper输出KV中的K,则该对象还需实现Comparable接口，因为因为MapReduce框中的Shuffle过程要求对key必须能排序！！排序内容后面说！@Override public int compareTo(CustomBean o) { // 自定义排序规则 return this.num > o.getNum() ? -1 : 1; }Writable接口案例需求统计每台智能音箱设备内容播放时长原始日志格式：001577c3 kar_890809 120.196.100.99 1116 954 200日志id 设备id appkey(合作硬件厂商) 网络ip 自有内容时长(秒) 第三方内容时长(秒) 网络状态码输出结果：001577c3 11160 9540 20700设备id 自有内容时长(秒) 第三方内容时长(秒) 总时长编写MapReduce程序创建SpeakBean对象package com.test.hdfs; import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; // 1 实现writable接口 public class SpeakBean implements Writable { private long selfDuration; private long thirdPartDuration; private long sumDuration; //2 反序列化时，需要反射调用空参构造函数，所以必须有 public SpeakBean() { } public SpeakBean(long selfDuration, long thirdPartDuration) { this.selfDuration = selfDuration; this.thirdPartDuration = thirdPartDuration; this.sumDuration=this.selfDuration+this.thirdPartDuration; } //3 写序列化方法 public void write(DataOutput out) throws IOException { out.writeLong(selfDuration); out.writeLong(thirdPartDuration); out.writeLong(sumDuration); } //4 反序列化方法 //5 反序列化方法读顺序必须和写序列化方法的写顺序必须一致 public void readFields(DataInput in) throws IOException { this.selfDuration = in.readLong(); this.thirdPartDuration = in.readLong(); this.sumDuration = in.readLong(); } // 6 编写toString方法，方便后续打印到文本 @Override public String toString() { return selfDuration + "\t" + thirdPartDuration + "\t" + sumDuration ; } public long getSelfDuration() { return selfDuration; } public void setSelfDuration(long selfDuration) { this.selfDuration = selfDuration; } public long getThirdPartDuration() { return thirdPartDuration; } public void setThirdPartDuration(long thirdPartDuration) { this.thirdPartDuration = thirdPartDuration; } public long getSumDuration() { return sumDuration; } public void setSumDuration(long sumDuration) { this.sumDuration = sumDuration; } public void set(long selfDuration, long thirdPartDuration) { this.selfDuration = selfDuration; this.thirdPartDuration = thirdPartDuration; this.sumDuration=this.selfDuration+this.thirdPartDuration; } }编写Mapper类package com.test.hdfs; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class SpeakDurationMapper extends Mapper<LongWritable, Text, Text, SpeakBean> { SpeakBean v = new SpeakBean(); Text k = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 1 获取一行 String line = value.toString(); // 2 切割字段 String[] fields = line.split("\t"); // 3 封装对象 // 取出设备id String deviceId = fields[1]; // 取出自有和第三方时长数据 long selfDuration = Long.parseLong(fields[fields.length - 3]); long thirdPartDuration = Long.parseLong(fields[fields.length - 2]); k.set(deviceId); v.set(selfDuration, thirdPartDuration); // 4 写出 context.write(k, v); } }编写Reducerpackage com.test.hdfs; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class SpeakDurationReducer extends Reducer<Text, SpeakBean, Text, SpeakBean> { @Override protected void reduce(Text key, Iterable<SpeakBean> values, Context context)throws IOException, InterruptedException { long self_Duration = 0; long thirdPart_Duration = 0; // 1 遍历所用bean，将其中的自有，第三方时长分别累加 for (SpeakBean sb : values) { self_Duration += sb.getSelfDuration(); thirdPart_Duration += sb.getThirdPartDuration(); } // 2 封装对象 SpeakBean resultBean = new SpeakBean(self_Duration, thirdPart_Duration); // 3 写出 context.write(key, resultBean); } }编写驱动package com.test.hdfs; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; 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 java.io.IOException; public class SpeakerDriver { public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException { // 输入输出路径需要根据自己电脑上实际的输入输出路径设置 args = new String[] { "e:/input/input", "e:/output1" }; // 1 获取配置信息，或者job对象实例 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 6 指定本程序的jar包所在的本地路径 job.setJarByClass(SpeakerDriver.class); // 2 指定本业务job要使用的mapper/Reducer业务类 job.setMapperClass(SpeakDurationMapper.class); job.setReducerClass(SpeakDurationReducer.class); // 3 指定mapper输出数据的kv类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(SpeakBean.class); // 4 指定最终输出的数据的kv类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(SpeakBean.class); // 5 指定job的输入原始文件所在目录 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行 boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }mr编程技巧总结结合业务设计Map输出的key和v，利用key相同则去往同一个reduce的特点！！map()方法中获取到只是一行文本数据尽量不做聚合运算reduce()方法的参数要清楚含义MapReduce原理分析MapTask运行机制详解MapTask流程详细步骤：首先，读取数据组件InputFormat（默认TextInputFormat）会通过getSplits方法对输入目录中文件进行逻辑切片规划得到splits，有多少个split就对应启动多少个MapTask。split与block的对应关系默认是一对一。将输入文件切分为splits之后，由RecordReader对象（默认LineRecordReader）进行读取，以\n作为分隔符，读取一行数据，返回<key，value>。Key表示每行首字符偏移值，value表示这一行文本内容。读取split返回<key,value>，进入用户自己继承的Mapper类中，执行用户重写的map函数。RecordReader读取一行这里调用一次。map逻辑完之后，将map的每条结果通过context.write进行collect数据收集。在collect中，会先对其进行分区处理，默认使用HashPartitioner。MapReduce提供Partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。接下来，会将数据写入内存，内存中这片区域叫做环形缓冲区，缓冲区的作用是批量收集map结果，减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然写入之前，key与value值都会被序列化成字节数组。1）环形缓冲区其实是一个数组，数组中存放着key、value的序列化数据和key、value的元数据信息，包括partition、key的起始位置、value的起始位置以及value的长度。环形结构是一个抽象概念。2）缓冲区是有大小限制，默认是100MB。当map task的输出结果很多时，就可能会撑爆内存，所以需要在一定条件下将缓冲区中的数据临时写入磁盘，然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为Spill，中文可译为溢写。这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8，也就是当缓冲区的数据已经达到阈值（buffer size spill percent = 100MB 0.8 = 80MB），溢写线程启动，锁定这80MB的内存，执行溢写过程。Maptask的输出结果还可以往剩下的20MB内存中写，互不影响。当溢写线程启动后，需要对这80MB空间内的key做排序(Sort)。排序是MapReduce模型默认的行为!1）如果job设置过Combiner，那么现在就是使用Combiner的时候了。将有相同key的key/value对的value加起来，减少溢写到磁盘的数据量。Combiner会优化MapReduce的中间结果，所以它在整个模型中会多次使用。2）那哪些场景才能使用Combiner呢？从这里分析，Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。Combiner的使用一定得慎重，如果用好，它对job执行效率有帮助，反之会影响reduce的最终结果。合并溢写文件：每次溢写会在磁盘上生成一个临时文件（写之前判断是否有combiner），如果map的输出结果真的很大，有多次这样的溢写发生，磁盘上相应的就会有多个临时文件存在。当整个数据处理结束之后开始对磁盘中的临时文件进行merge合并，因为最终的文件只有一个，写入磁盘，并且为这个文件提供了一个索引文件，以记录每个reduce对应数据的偏移量。至此map整个阶段结束!!MapTask的一些配置：官方参考地址：https://hadoop.apache.org/docs/r2.9.2/hadoop-mapreduce-client/hadoop-mapreduceclient-core/mapred-default.xmlMapTask的并行度MapTask并行度思考MapTask的并行度决定Map阶段的任务处理并发度，从而影响到整个Job的处理速度。思考：MapTask并行任务是否越多越好呢？哪些因素影响了MapTask并行度？MapTask并行度决定机制数据块：Block是HDFS物理上把数据分成一块一块。切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。切片机制源码阅读默认就是128M；MapTask并行度是不是越多越好呢？答案不是，如果一个文件仅仅比128M大一点点也被当成一个split来对待，而不是多个split。MR框架在并行运算的同时也会消耗更多资源，并行度越高资源消耗也越高，假设129M文件分为两个分片，一个是128M，一个是1M；对于1M的切片的Maptask来说，太浪费资源。思考：129M的文件在Hdfs存储的时候会不会切成两块？ReduceTask 工作机制Reduce大致分为copy、sort、reduce三个阶段，重点在前两个阶段。copy阶段包含一个eventFetcher来获取已完成的map列表，由Fetcher线程去copy数据，在此过程中会启动两个merge线程，分别为inMemoryMerger和onDiskMerger，分别将内存中的数据merge到磁盘和将磁盘中的数据进行merge。待数据copy完成之后，copy阶段就完成了，开始进行sort阶段，sort阶段主要是执行finalMerge操作，纯粹的sort阶段，完成之后就是reduce阶段，调用用户定义的reduce函数进行处理。详细步骤：Copy阶段，简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher)，通过HTTP方式请求maptask获取属于自己的文件。Merge阶段。这里的merge如map端的merge动作，只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活。merge有三种形式：内存到内存；内存到磁盘；磁盘到磁盘。默认情况下第一种形式不启用。当内存中的数据量到达一定阈值，就启动内存到磁盘的merge。与map 端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行，直到没有map端的数据时才结束，然后启动第三种磁盘到磁盘的merge方式生成最终的文件。合并排序。把分散的数据合并成一个大的数据后，还会再对合并后的数据排序。对排序后的键值对调用reduce方法，键相等的键值对调用一次reduce方法，每次调用会产生零个或者多个键值对，最后把这些输出的键值对写入到HDFS文件中。ReduceTask并行度ReduceTask的并行度同样影响整个Job的执行并发度和执行效率，但与MapTask的并发数由切片数决定不同，ReduceTask数量的决定是可以直接手动设置：// 默认值是1，手动设置为4 job.setNumReduceTasks(4);注意事项ReduceTask=0，表示没有Reduce阶段，输出文件数和MapTask数量保持一致；ReduceTask数量不设置默认就是一个，输出文件数量为1个；如果数据分布不均匀，可能在Reduce阶段产生倾斜；Shuffle机制map阶段处理的数据如何传递给reduce阶段，是MapReduce框架中最关键的一个流程，这个流程就叫shuffle。shuffle: 洗牌、发牌——（核心机制：数据分区，排序，分组，combine，合并等过程）MapReduce的分区与reduceTask的数量在MapReduce中，通过我们指定分区，会将同一个分区的数据发送到同一个reduce当中进行处理(默认是key相同去往同个分区)，例如我们为了数据的统计，我们可以把一批类似的数据发送到同一个reduce当中去，在同一个reduce当中统计相同类型的数据，如何才能保证相同key的数据去往同个reduce呢？只需要保证相同key的数据分发到同个分区即可。结合以上原理分析我们知道MR程序shuffle机制默认就是这种规则！！分区源码翻阅源码验证以上规则，MR程序默认使用的HashPartitioner，保证了相同的key去往同个分区！！自定义分区实际生产中需求变化多端，默认分区规则往往不能满足需求，需要结合业务逻辑来灵活控制分区规则以及分区数量！！如何制定自己需要的分区规则？具体步骤：1）自定义类继承Partitioner，重写getPartition()方法2）在Driver驱动中，指定使用自定义Partitioner3）在Driver驱动中，要根据自定义Partitioner的逻辑设置相应数量的ReduceTask数量。案例：需求：按照不同的appkey把记录输出到不同的分区中原始日志格式：001577c3 kar_890809 120.196.100.99 1116 954 200日志id 设备id appkey(合作硬件厂商) 网络ip 自有内容时长(秒) 第三方内容时长(秒) 网络状态码输出结果：根据appkey把不同厂商的日志数据分别输出到不同的文件中需求分析：面对业务需求，结合mr的特点，来设计map输出的kv,以及reduce输出的kv数据。一个ReduceTask对应一个输出文件，因为在shuffle机制中每个reduceTask拉取的都是某一个分区的数据，一个分区对应一个输出文件。结合appkey的前缀相同的特点，同时不能使用默认分区规则，而是使用自定义分区器，只要appkey前缀相同则数据进入同个分区。整体思路：Mapper：1）读取一行文本，按照制表符切分2）解析出appkey字段，其余数据封装为PartitionBean对象（实现序列化Writable接口）3）设计map()输出的kv,key-->appkey(依靠该字段完成分区)，PartitionBean对象作为Value输出Partition：自定义分区器，实现按照appkey字段的前缀来区分所属分区Reduce：reduce()正常输出即可，无需进行聚合操作Driver：1）在原先设置job属性的同时增加设置使用自定义分区器2）注意设置ReduceTask的数量（与分区数量保持一致）Mapperpackage com.test.mr.partition; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class PartitionMapper extends Mapper<LongWritable, Text, Text, PartitionBean> { PartitionBean bean = new PartitionBean(); Text text = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split("\t"); String appkey = fields[2]; bean.setId(fields[0]); bean.setDeviceId(fields[1]); bean.setAppkey(appkey); bean.setIp(fields[3]); bean.setSelfDuration(Long.parseLong(fields[4])); bean.setThirdPartDuration(Long.parseLong(fields[5])); bean.setStatus(fields[6]); text.set(appkey); context.write(text, bean); } }PartitionBeanpackage com.test.mr.partition; import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class PartitionBean implements Writable { //定义属性 private String id; //日志id private String deviceId;//设备id private String appkey;//appkey合作硬件厂商id private String ip;//ip地址 private Long selfDuration;//自有内容时长 private Long thirdPartDuration;//第三方内容时长 private String status;//状态码 public PartitionBean() { } public PartitionBean(String id, String deviceId, String appkey, String ip, Long selfDuration, Long thirdPartDuration, String status) { this.id = id; this.deviceId = deviceId; this.appkey = appkey; this.ip = ip; this.selfDuration = selfDuration; this.thirdPartDuration = thirdPartDuration; this.status = status; } public String getId() { return id; } public void setId(String id) { this.id = id; } public String getDeviceId() { return deviceId; } public void setDeviceId(String deviceId) { this.deviceId = deviceId; } public String getAppkey() { return appkey; } public void setAppkey(String appkey) { this.appkey = appkey; } public String getIp() { return ip; } public void setIp(String ip) { this.ip = ip; } public Long getSelfDuration() { return selfDuration; } public void setSelfDuration(Long selfDuration) { this.selfDuration = selfDuration; } public Long getThirdPartDuration() { return thirdPartDuration; } public void setThirdPartDuration(Long thirdPartDuration) { this.thirdPartDuration = thirdPartDuration; } public String getStatus() { return status; } public void setStatus(String status) { this.status = status; } //序列化 @Override public void write(DataOutput out) throws IOException { out.writeUTF(id); out.writeUTF(deviceId); out.writeUTF(appkey); out.writeUTF(ip); out.writeLong(selfDuration); out.writeLong(thirdPartDuration); out.writeUTF(status); } //反序列化 @Override public void readFields(DataInput in) throws IOException { this.id = in.readUTF(); this.deviceId=in.readUTF(); this.appkey=in.readUTF(); this.ip=in.readUTF(); this.selfDuration=in.readLong(); this.thirdPartDuration=in.readLong(); this.status=in.readUTF(); } @Override public String toString() { return id + '\t' + deviceId + '\t' + appkey + '\t' + ip + '\t' + selfDuration +'\t'+ thirdPartDuration + '\t' + status ; } }CustomPartitionerpackage com.test.mr.partition; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; public class CustomPartitioner extends Partitioner<Text,PartitionBean> { @Override public int getPartition(Text text, PartitionBean partitionBean, int numPartitions) { int partition=0; String appkey = text.toString(); if(appkey.equals("kar")){ partition=1; }else if(appkey.equals("pandora")){ partition=2; }else{ partition=0; } return partition; } }PartitionReducerpackage com.test.mr.partition; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class PartitionReducer extends Reducer<Text, PartitionBean, NullWritable, PartitionBean> { @Override protected void reduce(Text key, Iterable<PartitionBean> values, Context context) throws IOException, InterruptedException { for (PartitionBean bean : values) { context.write(NullWritable.get(), bean); } } }PartitionDriverpackage com.test.mr.partition; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; 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 java.io.IOException; public class PartitionDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(PartitionDriver.class); job.setMapperClass(PartitionMapper.class); job.setReducerClass(PartitionReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(PartitionBean.class); job.setOutputKeyClass(NullWritable.class); job.setOutputValueClass(PartitionBean.class); job.setPartitionerClass(CustomPartitioner.class); job.setNumReduceTasks(5); FileInputFormat.setInputPaths(job, new Path("e:/speak.data")); FileOutputFormat.setOutputPath(job, new Path("e:/partition/output")); final boolean flag = job.waitForCompletion(true); System.exit(flag ? 0 : 1); } }总结1）自定义分区器时最好保证分区数量与reduceTask数量保持一致；2）如果分区数量不止1个，但是reduceTask数量1个，此时只会输出一个文件。3）如果reduceTask数量大于分区数量，但是输出多个空文件4）如果reduceTask数量小于分区数量，有可能会报错。MapReduce中的Combinercombiner运行机制：Combiner是MR程序中Mapper和Reducer之外的一种组件Combiner组件的父类就是ReducerCombiner和reducer的区别在于运行的位置Combiner是在每一个maptask所在的节点运行;Combiner的意义就是对每一个maptask的输出进行局部汇总，以减小网络传输量。Combiner能够应用的前提是不能影响最终的业务逻辑，此外，Combiner的输出kv应该跟reducer的输入kv类型要对应起来。举例说明：假设一个计算平均值的MR任务Map阶段有2个MapTaskMapTask1输出数据：10,5,15 如果使用Combiner:(10+5+15)/3=10MapTask2输出数据：2,6 如果使用Combiner:(2+6)/2=4Reduce阶段汇总（10+4）/2=7而正确结果应该是（10+5+15+2+6）/5=7.6自定义Combiner实现步骤：自定义一个Combiner继承Reducer，重写Reduce方法在驱动(Driver)设置使用Combiner（默认是不适用Combiner组件）改造WordCount程序package com.test.mr.wc; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.io.Text; import java.io.IOException; public class WordCountCombiner extends Reducer<Text, IntWritable,Text,IntWritable> { IntWritable total = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { // 遍历key对应的values，然后累加结果 int sum = 0; for (IntWritable value : values) { int i = value.get(); sum += 1; } // 直接输出当前key对应的sum值，结果就是单词出现的总次数 total.set(sum); context.write(key, total); } }在驱动(Driver)设置使用Combinerjob.setCombinerClass(WordcountCombiner.class);验证结果，观察程序运行日志如果直接使用WordCountReducer作为Combiner使用是否可以？可以。MapReduce中的排序排序是MapReduce框架中最重要的操作之一。MapTask和ReduceTask均会对数据按照key进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要。默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。MapTask它会将处理的结果暂时放到环形缓冲区中，当环形缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次快速排序，并将这些有序数据溢写到磁盘上，溢写完毕后，它会对磁盘上所有文件进行归并排序。ReduceTask当所有数据拷贝完毕后，ReduceTask统-对内存和磁盘上的所有数据进行一次归并排序。部分排序.MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部有序。全排序最终输出结果只有一个文件，且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了MapReduce所提供的并行架构。辅助排序: (GroupingComparator分组)在Reduce端对key进行分组。应用于:在接收的key为bean对象时，想让一个或几个字段相同(全部字段比较不相同)的key进入到同一个reduce方法时，可以采用分组排序。4.二次排序.在自定义排序过程中，如果compareTo中的判断条件为两个即为二次排序。全排序WritableComparableBean对象如果作为Map输出的key时，需要实现WritableComparable接口并重写compareTo方法指定排序规则基于统计的播放时长案例的输出结果对总时长进行排序实现全局排序只能设置一个ReduceTask!!播放时长案例输出结果：00fdaf3 33180 33420 00fdaf3 6660000wersa4 30689 35191 00wersa4 658800a0fe2 43085 44254 0a0fe2 873390ad0s7 31702 29183 0ad0s7 608850sfs01 31883 29101 0sfs01 60984a00df6s 33239 36882 a00df6s 70121adfd00fd5 30727 31491 adfd00fd5 62218需求分析：如何设计map()方法输出的key和value？MR框架中shuffle阶段的排序是默认行为，不管你是否需要都会进行排序。key:把所有字段封装成为一个bean对象，并且指定bean对象作为key输出，如果作为key输出，需要实现排序接口，指定自己的排序规则；具体步骤Mapper读取结果文件，按照制表符进行切分解析出相应字段封装为SpeakBeanSpeakBean实现WritableComparable接口重写compareTo方法map()方法输出kv;key-->SpeakBean,value-->NullWritable.get()Reducer循环遍历输出Mapper代码package com.test.mr.WritableComparable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class SortMapper extends Mapper<LongWritable, Text, SpeakBeanSort, NullWritable> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] arr = value.toString().split("\t"); SpeakBeanSort bean = new SpeakBeanSort(Long.parseLong(arr[1]), Long.parseLong(arr[2]), arr[0]); context.write(bean, NullWritable.get()); } }Reducerpackage com.test.mr.WritableComparable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class SortReducer extends Reducer<SpeakBeanSort, NullWritable, NullWritable, SpeakBeanSort> { @Override protected void reduce(SpeakBeanSort key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { for (NullWritable value : values) { context.write(value, key); } } }Bean对象实现WritableComparable接口package com.test.mr.WritableComparable; import org.apache.hadoop.io.WritableComparable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class SpeakBeanSort implements WritableComparable<SpeakBeanSort> { //定义属性 private Long selfDuration;//自有内容时长 private Long thirdPartDuration;//第三方内容时长 private String deviceId;//设备id private Long sumDuration;//总时长 //准备一个空参构造 public SpeakBeanSort() { } //序列化方法:就是把内容输出到网络或者文本中 @Override public void write(DataOutput out) throws IOException { out.writeLong(selfDuration); out.writeLong(thirdPartDuration); out.writeUTF(deviceId); out.writeLong(sumDuration); } //有参构造 public SpeakBeanSort(Long selfDuration, Long thirdPartDuration, String deviceId) { this.selfDuration = selfDuration; this.thirdPartDuration = thirdPartDuration; this.deviceId = deviceId; this.sumDuration = this.selfDuration + this.thirdPartDuration; } //反序列化方法 @Override public void readFields(DataInput in) throws IOException { this.selfDuration = in.readLong();//自由时长 this.thirdPartDuration = in.readLong();//第三方时长 this.deviceId = in.readUTF();//设备id this.sumDuration = in.readLong();//总时长 } public Long getSelfDuration() { return selfDuration; } public void setSelfDuration(Long selfDuration) { this.selfDuration = selfDuration; } public Long getThirdPartDuration() { return thirdPartDuration; } public void setThirdPartDuration(Long thirdPartDuration) { this.thirdPartDuration = thirdPartDuration; } public String getDeviceId() { return deviceId; } public void setDeviceId(String deviceId) { this.deviceId = deviceId; } public Long getSumDuration() { return sumDuration; } public void setSumDuration(Long sumDuration) { this.sumDuration = sumDuration; } //为了方便观察数据，重写toString()方法 @Override public String toString() { return selfDuration + "\t" + thirdPartDuration + "\t" + deviceId + "\t" + sumDuration; } //重写compareTo方法，此处是比较一个字段如果比较2个字段就是二次排序 @Override public int compareTo(SpeakBeanSort o) { int result; // 按照总流量大小，倒序排列 if (sumDuration > o.getSumDuration()) { result = -1; }else if (sumDuration < o.getSumDuration()) { result = 1; }else { result = 0; } return result; } }Driverpackage com.test.mr.WritableComparable; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class SortDriver { public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException { // 1 获取配置信息，或者job对象实例 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 2 指定本程序的jar包所在的本地路径 job.setJarByClass(SortDriver.class); // 3 指定本业务job要使用的mapper/Reducer业务类 job.setMapperClass(SortMapper.class); job.setReducerClass(SortReducer.class); // 4 指定mapper输出数据的kv类型 job.setMapOutputKeyClass(SpeakBeanSort.class); job.setMapOutputValueClass(NullWritable.class); // 5 指定最终输出的数据的kv类型 job.setOutputKeyClass(NullWritable.class); job.setOutputValueClass(SpeakBeanSort.class); // 6 指定job的输入原始文件所在目录 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行 boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }总结自定义对象作为Map的key输出时，需要实现WritableComparable接口，排序：重写compareTo()方法，序列以及反序列化方法再次理解reduce()方法的参数；reduce()方法是map输出的kv中key相同的kv中的v组成一个集合调用一次reduce()方法，选择遍历values得到所有的key.默认reduceTask数量是1个；对于全局排序需要保证只有一个reduceTask分区排序（默认的分区规则，区内有序）GroupingComparatorGroupingComparator是mapreduce当中reduce端的一个功能组件，主要的作用是决定哪些数据作为一组，调用一次reduce的逻辑，默认是每个不同的key，作为多个不同的组，每个组调用一次reduce逻辑，我们可以自定义GroupingComparator实现不同的key作为同一个组，调用一次reduce逻辑。需求原始数据订单id商品id成交金额Order_0000001Pdt_01222.8Order_0000001Pdt_0525.8Order_0000002Pdt_03522.8Order_0000002Pdt_04122.4Order_0000002Pdt_05722.4Order_0000003Pdt_01232.8需要求出每一个订单中成交金额最大的一笔交易。实现思路Mapper1）读取一行文本数据，切分出每个字段；2）订单id和金额封装为一个Bean对象，Bean对象的排序规则指定为先按照订单Id排序，订单Id相等再按照金额降序排；3）map()方法输出kv;key-->bean对象，value-->NullWritable.get()；Shuffle1）指定分区器，保证相同订单id的数据去往同个分区（自定义分区器）2）指定GroupingComparator，分组规则指定只要订单Id相等则认为属于同一组；Reduce每个reduce()方法写出一组key的第一个参考代码OrderBeanOrderBean定义两个字段，一个字段是orderId，第二个字段是金额（注意金额一定要使用Double或者DoubleWritable类型，否则没法按照金额顺序排序）排序规则指定为先按照订单Id排序，订单Id相等再按照金额降序排！！package com.test.mr.group; import org.apache.hadoop.io.WritableComparable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class OrderBean implements WritableComparable<OrderBean> { private String orderId; private Double price; @Override public int compareTo(OrderBean o) { //比较订单id的排序顺序 int i = this.orderId.compareTo(o.orderId); if(i == 0) { //如果订单id相同，则比较金额，金额大的排在前面 i = -this.price.compareTo(o.price); } return i; } @Override public void write(DataOutput out) throws IOException { out.writeUTF(orderId); out.writeDouble(price); } @Override public void readFields(DataInput in ) throws IOException { this.orderId = in .readUTF(); this.price = in .readDouble(); } public OrderBean() { } public OrderBean(String orderId, Double price) { this.orderId = orderId; this.price = price; } public String getOrderId() { return orderId; } public void setOrderId(String orderId) { this.orderId = orderId; } public Double getPrice() { return price; } public void setPrice(Double price) { this.price = price; } @Override public String toString() { return orderId + "\t" + price; } }自定义分区器保证ID相同的订单去往同个分区最终去往同一个Reduce中package com.test.mr.group; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Partitioner; public class CustomPartitioner extends Partitioner<OrderBean, NullWritable> { @Override public int getPartition(OrderBean orderBean, NullWritable nullWritable, int i) { //自定义分区，将相同订单id的数据发送到同一个reduce里面去 return (orderBean.getOrderId().hashCode() & Integer.MAX_VALUE) % i; } }自定义GroupingComparator保证id相同的订单进入一个分组中，进入分组的数据已经是按照金额降序排序。reduce()方法取出第一个即是金额最高的交易package com.test.mr.group; import org.apache.hadoop.io.WritableComparable; import org.apache.hadoop.io.WritableComparator; public class CustomGroupingComparator extends WritableComparator { //将我们自定义的OrderBean注册到我们自定义的CustomGroupIngCompactor当中来 //表示我们的分组器在分组的时候，对OrderBean这一种类型的数据进行分组 //传入作为key的bean的class类型，以及制定需要让框架做反射获取实例对象 public CustomGroupingComparator() { super(OrderBean.class, true); } @Override public int compare(WritableComparable a, WritableComparable b) { OrderBean first = (OrderBean) a; OrderBean second = (OrderBean) b; final int i = first.getOrderId().compareTo(second.getOrderId()); if (i == 0) { System.out.println(first.getOrderId() + "----" + second.getOrderId()); } return i; } }Mapperpackage com.test.mr.group; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class GroupMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //读取一行文本，然后切分 String[] arr = value.toString().split("t"); OrderBean bean = new OrderBean(arr[0], Double.parseDouble(arr[2])); context.write(bean, NullWritable.get()); } }Reducerpackage com.test.mr.group; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class GroupReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> { @Override protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { context.write(key, NullWritable.get()); } }Driverpackage com.test.mr.group; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class GroupDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { // 1 获取配置信息，或者job对象实例 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 2 指定本程序的jar包所在的本地路径 job.setJarByClass(GroupDriver.class); // 3 指定本业务job要使用的mapper/Reducer业务类 job.setMapperClass(GroupMapper.class); job.setReducerClass(GroupReducer.class); // 4 指定mapper输出数据的kv类型 job.setMapOutputKeyClass(OrderBean.class); job.setMapOutputValueClass(NullWritable.class); // 5 指定最终输出的数据的kv类型 job.setOutputKeyClass(OrderBean.class); job.setOutputValueClass(NullWritable.class); // 6 指定job的输入原始文件所在目录 FileInputFormat.setInputPaths(job, new Path("E:\teach\hadoop框架\资料\data\GroupingComparator")); FileOutputFormat.setOutputPath(job, new Path("E:\group-out")); // 7 指定分区器，指定分组比较器，设置reducetask数量 job.setPartitionerClass(CustomPartitioner.class); job.setGroupingComparatorClass(CustomGroupingComparator.class); job.setNumReduceTasks(2); // 8 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行 Boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }MapReduce读取和输出数据InputFormat运行MapReduce程序时，输入的文件格式包括:基于行的日志文件、二进制格式文件、数据库表等。那么，针对不同的数据类型，MapReduce是如何读取这些数据的呢？InputFormat是MapReduce框架用来读取数据的类。InputFormat常见子类包括:TextInputFormat （普通文本文件，MR框架默认的读取实现类型）KeyValueTextInputFormat（读取一行文本数据按照指定分隔符，把数据封装为kv类型）NLineInputF ormat(读取数据按照行数进行划分分片)CombineTextInputFormat(合并小文件，避免启动过多MapTask任务)自定义InputFormatCombineTextInputFormat案例MR框架默认的TextInputFormat切片机制按文件划分切片，文件无论多小，都是单独一个切片，然后由一个MapTask处理，如果有大量小文件，就对应的会生成并启动大量的 MapTask，而每个MapTask处理的数据量很小大量时间浪费在初始化资源启动收回等阶段，这种方式导致资源利用率不高。CombineTextInputFormat用于小文件过多的场景，它可以将多个小文件从逻辑上划分成一个切片，这样多个小文件就可以交给一个MapTask处理，提高资源利用率。需求：将输入数据中的多个小文件合并为一个切片处理使用方式：运行WordCount案例，准备多个小文件// 如果不设置InputFormat，它默认用的是TextInputFormat.class job.setInputFormatClass(CombineTextInputFormat.class); //虚拟存储切片最大值设置4m CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);验证切片数量的变化CombineTextInputFormat切片原理：切片生成过程分为两部分：虚拟存储过程和切片过程假设设置setMaxInputSplitSize值为4M四个小文件：1.txt→2M；2.txt→7M；3.txt→0.3M；4.txt→8.2M虚拟存储过程：把输入目录下所有文件大小，依次和设置的setMaxInputSplitSize值进行比较，如果不大于设置的最大值，逻辑上划分一个块。如果输入文件大于设置的最大值且大于两倍，那么以最大值切割一块；当剩余数据大小超过设置的最大值且不大于最大值2倍，此时将文件均分成2个虚拟存储块（防止出现太小切片）。比如如setMaxInputSplitSize值为4M，输入文件大小为8.02M，则先逻辑上分出一个4M的块。剩余的大小为4.02M，如果按照4M逻辑划分，就会出现0.02M的非常小的虚拟存储文件，所以将剩余的4.02M文件切分成（2.01M和2.01M）两个文件。1.txt→2M;2M<4M;一个块；2.txt→7M;7M>4M,但是不大于两倍，均匀分成两块；两块：每块3.5M；3.txt→0.3M;0.3<4M ,0.3M<4M ,一个块4.txt→8.2M;大于最大值且大于两倍；一个4M的块，剩余4.2M分成两块，每块2.1M所有块信息：2M，3.5M，3.5M，0.3M，4M，2.1M，2.1M 共7个虚拟存储块。切片过程判断虚拟存储的文件大小是否大于setMaxInputSplitSize值，大于等于则单独形成一个切片。如果不大于则跟下一个虚拟存储文件进行合并，共同形成一个切片。按照之前输入文件：有4个小文件大小分别为2M、7M、0.3M以及8.2M这四个小文件，则虚拟存储之后形成7个文件块，大小分别为：2M，3.5M，3.5M，0.3M，4M，2.1M，2.1M最终会形成3个切片，大小分别为：（2+3.5）M，（3.5+0.3+4）M，（2.1+2.1）M注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。自定义InputFormatHDFS还是MapReduce，在处理小文件时效率都非常低，但又难免面临处理大量小文件的场景，此时，就需要有相应解决方案。可以自定义InputFormat实现小文件的合并。需求：将多个小文件合并成一个SequenceFile文件（SequenceFile文件是Hadoop用来存储二进制形式的key-value对的文件格式），SequenceFile里面存储着多个文件，存储的形式为文件路径+名称为key，文件内容为value。结果：得到一个合并了多个小文件的SequenceFile文件整体思路：定义一个类继承FileInputFormat（TextInputFormat是它的子类）重写isSplitable()指定为不可切分；重写createRecordReader()方法，创建自己的RecorderReader（实现自定义读取数据）对象改变默认读取数据方式，实现一次读取一个完整文件作为kv输出；Driver指定使用的InputFormat类型代码参考：自定义InputFormatpackage com.test.mr.inputformat; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.InputSplit; import org.apache.hadoop.mapreduce.JobContext; import org.apache.hadoop.mapreduce.RecordReader; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import java.io.IOException; public class CustomFileInputformat extends FileInputFormat<Text, BytesWritable>{ //文件不可切分 @Override protected Boolean isSplitable(JobContext context, Path filename) { return false; } //获取自定义RecordReader对象用来读取数据 @Override public RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException { CustomRecordReader recordReader = new CustomRecordReader(); recordReader.initialize(split, context); return recordReader; } }自定义RecordReaderpackage com.test.mr.inputformat; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.InputSplit; import org.apache.hadoop.mapreduce.RecordReader; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.hadoop.mapreduce.lib.input.FileSplit; import java.io.IOException; public class CustomRecordReader extends RecordReader<Text, BytesWritable> { private Configuration configuration; //切片 private FileSplit split; //是否读取到内容的标识符 private Boolean isProgress = true; //输出的kv private BytesWritable value = new BytesWritable(); private Text k = new Text(); @Override public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException { //获取到文件切片以及配置文件对象 this.split = (FileSplit) split; configuration = context.getConfiguration(); } @Override public Boolean nextKeyValue() throws IOException, InterruptedException { if (isProgress) { // 1 定义缓存区 byte[] contents = new byte[(int) split.getLength()]; FileSystem fs = null; FSDataInputStream fis = null; try { // 2 获取文件系统 Path path = split.getPath(); fs = path.getFileSystem(configuration); // 3 读取数据 fis = fs.open(path); // 4 读取文件内容 IOUtils.readFully(fis, contents, 0, contents.length); // 5 输出文件内容 value.set(contents, 0, contents.length); // 6 获取文件路径及名称 String name = split.getPath().toString(); // 7 设置输出的key值 k.set(name); } catch (Exception e) { } finally { IOUtils.closeStream(fis); } isProgress = false; return true; } return false; } @Override public Text getCurrentKey() throws IOException, InterruptedException { //返回key return k; } @Override public BytesWritable getCurrentValue() throws IOException, InterruptedException { //返回value return value; } @Override public float getProgress() throws IOException, InterruptedException { return 0; } @Override public void close() throws IOException { } }Mapperpackage com.test.mr.inputformat; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class SequenceFileMapper extends Mapper<Text, BytesWritable,Text,BytesWritable> { @Override protected void map(Text key, BytesWritable value, Context context) throws IOException, InterruptedException { //读取内容直接输出 context.write(key, value); } }Reducerpackage com.test.mr.inputformat; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class SequenceFileReducer extends Reducer<Text, BytesWritable, Text, BytesWritable> { @Override protected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException { //输出value值，其中只有一个BytesWritable 所以直接next取出即可 context.write(key, values.iterator().next()); } }Driverpackage com.test.mr.inputformat; import com.test.mr.partition.*; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.BytesWritable; 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 org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; import java.io.IOException; public class SequenceFileDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(SequenceFileDriver.class); job.setMapperClass(SequenceFileMapper.class); job.setReducerClass(SequenceFileReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(BytesWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(BytesWritable.class); job.setInputFormatClass(CustomFileInputformat.class); job.setOutputFormatClass(SequenceFileOutputFormat.class); FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); Boolean flag = job.waitForCompletion(true); System.exit(flag ? 0 : 1); } }验证输出结果OutputFormatOutputFormat:是MapReduce输出数据的基类，所有MapReduce的数据输出都实现了OutputFormat抽象类。下面介绍几种常见的OutputFormat子类TextOutputFormat默认的输出格式是TextOutputFormat，它把每条记录写为文本行。它的键和值可以是任意类型，因为TextOutputFormat调用toString()方 法把它们转换为字符串。SequenceFileOutputFormat将SequenceFileOutputFormat输出作为后续MapReduce任务的输入，这是一种好的输出格式，因为它的格式紧凑，很容易被压缩。自定义OutputFormat需求分析：要在一个MapReduce程序中根据数据的不同输出两类结果到不同目录，这类输出需求可以通过自定义OutputFormat来实现。实现步骤：自定义一个类继承FileOutputFormat。改写RecordWriter，改写输出数据的方法write()。需求：网络请求日志数据http://www.baidu.comhttp://www.google.comhttp://cn.bing.comhttps://www.zhaopin.comhttp://www.lagou.comhttps://www.liepin.comhttp://www.sohu.comhttp://www.sina.comhttp://www.sin2a.comhttp://www.sin2desa.comhttp://www.sindsafa.com输出结果：lagou.log文件，内容为：http://www.lagou.comother.log文件，内容为：http://www.baidu.comhttp://www.google.comhttp://cn.bing.comhttps://www.zhaopin.comhttps://www.liepin.comhttp://www.sohu.comhttp://www.sina.comhttp://www.sin2a.comhttp://www.sin2desa.comhttp://www.sindsafa.com参考代码：Mapperpackage com.test.mr.outputformat; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class OutputMapper extends Mapper<LongWritable, Text,Text, NullWritable>{ @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { context.write(value, NullWritable.get()); } }Reducerpackage com.test.mr.outputformat; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class OutputReducer extends Reducer<Text, NullWritable,Text,NullWritable>{ @Override protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { context.write(key,NullWritable.get() ); } }OutputFormatpackage com.test.mr.outputformat; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.RecordWriter; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class CustomOutputFormat extends FileOutputFormat<Text, NullWritable> { @Override public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException { //获取文件系统对象 FileSystem fs = FileSystem.get(context.getConfiguration()); //指定输出数据的文件 Path lagouPath = new Path("e:/lagou.log"); Path otherLog = new Path("e:/other.log"); //获取输出流 FSDataOutputStream lagouOut = fs.create(lagouPath); FSDataOutputStream otherOut = fs.create(otherLog); return new CustomWriter(lagouOut, otherOut); } }RecordWriterpackage com.test.mr.outputformat; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.RecordWriter; import org.apache.hadoop.mapreduce.TaskAttemptContext; import java.io.IOException; public class CustomWriter extends RecordWriter<Text, NullWritable> { private FSDataOutputStream lagouOut; private FSDataOutputStream otherOut; public CustomWriter(FSDataOutputStream lagouOut,FSDataOutputStream otherOut){ this.lagouOut=lagouOut; this.otherOut=otherOut; } @Override public void write(Text key, NullWritable value) throws IOException, InterruptedException { // 判断是否包含“lagou”输出到不同文件 if (key.toString().contains("lagou")) { lagouOut.write(key.toString().getBytes()); lagouOut.write("rn".getBytes()); } else { otherOut.write(key.toString().getBytes()); otherOut.write("rn".getBytes()); } } @Override public void close(TaskAttemptContext context) throws IOException, InterruptedException { IOUtils.closeStream(lagouOut); IOUtils.closeStream(otherOut); } }Driverpackage com.test.mr.outputformat; import com.test.mr.partition.*; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; 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 java.io.IOException; public class OutputDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(OutputDriver.class); job.setMapperClass(OutputMapper.class); job.setReducerClass(OutputReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); job.setOutputFormatClass(CustomOutputFormat.class); FileInputFormat.setInputPaths(job, new Path("e:/click_log.data")); // outputformat继承自fileoutputformat 而fileoutputformat要输出一个_SUCCESS文件，所以，在这还得指定一个输出目录 FileOutputFormat.setOutputPath(job, new Path("e:/click_log/output")); Boolean flag = job.waitForCompletion(true); System.exit(flag ? 0 : 1); } }验证结果是否已把数据分别输出到不同的目录中shuffle阶段数据的压缩机制hadoop当中支持的压缩算法数据压缩有两大好处，节约磁盘空间，加速数据在网络和磁盘上的传输。我们可以使用 hadoop checknative 命令来查看我们编译之后的hadoop支持的各种压缩，如果出现openssl为false，那么就在线安装一下依赖包。安装opensslyum install -y openssl-devel压缩格式hadoop自带算法文件扩展名是否可切分换成压缩格式后，原来的程序是否需要修改DEFLATE是，直接使用DEFLATE.deflate否和文本处理一样，不需要修改Gzip是，直接使用DEFLATE.gz否和文本处理一样，不需要修改bzip2是，直接使用bzip2.bz2是和文本处理一样，不需要修改LZO否，需要安装LZO.lzo是需要建索引，还需要指定输入格式Snappy否，需要安装Snappy.snappy否和文本处理一样，不需要修改为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器压缩格式对应的编码/解码器DEFLATEorg.apache.hadoop.io.compress.DefaultCodecgziporg.apache.hadoop.io.compress.GzipCodecbzip2org.apache.hadoop.io.compress.BZip2CodecLZOcom.hadoop.compression.lzo.LzopCodecSnappyorg.apache.hadoop.io.compress.SnappyCodec常见压缩方式对比分析压缩算法原始文件大小压缩后的文件大小压缩速度解压缩速度gzip8.3GB1.8GB17.5MB/s58MB/sbzip28.3GB1.1GB2.4MB/s9.5MB/sLZO-bset8.3GB2GB4MB/s60.6MB/sLZO8.3GB2.9GB49.3MB/S74.6MB/s压缩位置Map输入端压缩此处使用压缩文件作为Map的输入数据，无需显示指定编解码方式，Hadoop会自动检查文件扩展名，如果压缩方式能够匹配，Hadoop就会选择合适的编解码方式对文件进行压缩和解压。Map输出端压缩Shuffle是Hadoop MR过程中资源消耗最多的阶段，如果有数据量过大造成网络传输速度缓慢，可以考虑使用压缩Reduce端输出压缩输出的结果数据使用压缩能够减少存储的数据量，降低所需磁盘的空间，并且作为第二个MR的输入时可以复用压缩。压缩配置方式在驱动代码中通过Configuration直接设置使用的压缩方式，可以开启Map输出和Reduce输出压缩// 设置map阶段压缩 Configuration configuration = new Configuration(); configuration.set("mapreduce.map.output.compress","true"); configuration.set("mapreduce.map.output.compress.codec","org.apache.hadoop.io.compress.SnappyCodec"); // 设置reduce阶段的压缩 configuration.set("mapreduce.output.fileoutputformat.compress","true"); configuration.set("mapreduce.output.fileoutputformat.compress.type","RECORD"); configuration.set("mapreduce.output.fileoutputformat.compress.codec","org.apache.hadoop.io.compress.SnappyCodec");配置mapred-site.xml(修改后分发到集群其它节点，重启Hadoop集群)，此种方式对运行在集群的所有MR任务都会执行压缩。<property> <name>mapreduce.output.fileoutputformat.compress</name> <value>true</value> </property> <property> <name>mapreduce.output.fileoutputformat.compress.type</name> <value>RECORD</value> </property> <property> <name>mapreduce.output.fileoutputformat.compress.codec</name> <value>org.apache.hadoop.io.compress.SnappyCodec</value> </property>压缩案例需求：使用snappy压缩方式压缩WordCount案例的输出结果数据具体实现：在驱动代码中添加压缩配置configuration.set("mapreduce.output.fileoutputformat.compress","true"); configuration.set("mapreduce.output.fileoutputformat.compress.type","RECORD"); configuration.set("mapreduce.output.fileoutputformat.compress.codec","org.apache.hadoop.io.compress.SnappyCodec");重新打成jar包，提交集群运行，验证输出结果是否已进行了snappy压缩