MapReduce编程模型

MapReduce编程模型

1. 理解MapReduce编程模型
2. 独立完成一个MapReduce程序并运行成功
3. 了解MapReduce工程流程
4. 掌握并描述出shuffle全过程（面试）
5. 独立编写课堂及作业中的MR程序
6. 理解并解决数据倾斜

1. MapReduce编程模型

• Hadoop架构图

  Hadoop由HDFS分布式存储、MapReduce分布式计算、Yarn资源调度三部分组成
  

• MapReduce是采用一种分而治之的思想设计出来的分布式计算框架

• MapReduce由两个阶段组成：

  • Map阶段（切分成一个个小的任务）
  • Reduce阶段（汇总小任务的结果）

• 那什么是分而治之呢？

  • 比如一复杂、计算量大、耗时长的的任务，暂且称为“大任务”；
  • 此时使用单台服务器无法计算或较短时间内计算出结果时，可将此大任务切分成一个个小的任务，小任务分别在不同的服务器上并行的执行
  • 最终再汇总每个小任务的结果

1.1 Map阶段

• map阶段有一个关键的map()函数；
• 此函数的输入是键值对
• 输出是一系列键值对，输出写入本地磁盘。

1.2 Reduce阶段

• reduce阶段有一个关键的函数reduce()函数

• 此函数的输入也是键值对（即map的输出（kv对））

• 输出也是一系列键值对，结果最终写入HDFS

1.3 Map&Reduce

2. MapReduce编程示例

• 以MapReduce的词频统计为例：统计一批英文文章当中，每个单词出现的总次数

2.1 MapReduce原理图

• Map阶段
  • 假设MR的输入文件“Gone With The Wind”有三个block；block1、block2、block3
  • MR编程时，每个block对应一个分片split
  • 每一个split对应一个map任务（map task）
  • 如图共3个map任务（map1、map2、map3）；这3个任务的逻辑一样，所以以第一个map任务（map1）为例分析
  • map1读取block1的数据；一次读取block1的一行数据；
    • 产生键值对(key/value)，作为map()的参数传入，调用map()；
    • 假设当前所读行是第一行
    • 将当前所读行的行首相对于当前block开始处的字节偏移量作为key（0）
    • 当前行的内容作为value（Dear Bear River）
  • map()内
    • (按需求，写业务代码)，将value当前行内容按空格切分，得到三个单词Dear | Bear | River
    • 将每个单词变成键值对，输出出去(Dear, 1) | (Bear, 1) | (River, 1)；最终结果写入map任务所在节点的本地磁盘中（内里还有细节，讲到shuffle时，再细细展开）
    • block的第一行的数据被处理完后，接着处理第二行；逻辑同上
    • 当map任务将当前block中所有的数据全部处理完后，此map任务即运行结束
  • 其它的每一个map任务都是如上逻辑，不再赘述
• Reduce阶段
  • reduce任务（reduce task）的个数由自己写的程序编程指定，main()内的job.setNumReduceTasks(4)指定reduce任务是4个（reduce1、reduce2、reduce3、reduce4）
  • 每一个reduce任务的逻辑一样，所以以第一个reduce任务（reduce1）为例分析
  • map1任务完成后，reduce1通过网络，连接到map1，将map1输出结果中属于reduce1的分区的数据，通过网络获取到reduce1端（拷贝阶段）
  • 同样也如此连接到map2、map3获取结果
  • 最终reduce1端获得4个(Dear, 1)键值对；由于key键相同，它们分到同一组；
  • 4个(Dear, 1)键值对，转换成[Dear, Iterable(1, 1, 1, )]，作为两个参数传入reduce()
  • 在reduce()内部，计算Dear的总数为4，并将(Dear, 4)作为键值对输出
  • 每个reduce任务最终输出文件（内里还有细节，讲到shuffle时，再细细展开），文件写入到HDFS

2.2 MR中key的作用

• MapReduce编程中，key有特殊的作用

  • ①数据中，若要针对某个值进行分组、聚合时，需将此值作为MR中的reduce的输入的key

  • 如当前的词频统计例子，按单词进行分组，每组中对出现次数做聚合（计算总和）；所以需要将每个单词作为reduce输入的key，MapReduce框架自动按照单词分组，进而求出每组即每个单词的总次数
    

  • ②另外，key还具有可排序的特性，因为MR中的key类需要实现WritableComparable接口；而此接口又继承Comparable接口（可查看源码）

  • MR编程时，要充分利用以上两点；结合实际业务需求，设置合适的key

    
    

2.3 创建MAVEN工程

所有编程操作，在hadoop集群某节点的IDEA中完成

• 使用IDEA创建maven工程
• pom文件参考提供的pom.xml，主要用到的dependencies有

    <properties>
        <cdh.version>2.6.0-cdh5.14.2</cdh.version>
    </properties>

    <repositories>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>

    <dependencies>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.6.0-mr1-cdh5.14.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>${cdh.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>${cdh.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>${cdh.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/junit/junit -->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.testng</groupId>
            <artifactId>testng</artifactId>
            <version>RELEASE</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.38</version>
            <scope>compile</scope>
        </dependency>
    </dependencies>

2.4 MR参考代码

• 创建包com.kaikeba.hadoop.wordcount

• 在包中创建自定义mapper类、自定义reducer类、包含main类

2.4.1 Mapper代码

package com.kaikeba.hadoop.wordcount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * 类Mapper<LongWritable, Text, Text, IntWritable>的四个泛型分别表示
 * map方法的输入的键的类型kin、值的类型vin；输出的键的类型kout、输出的值的类型vout
 * kin指的是当前所读行行首相对于split分片开头的字节偏移量,所以是long类型，对应序列化类型LongWritable
 * vin指的是当前所读行，类型是String，对应序列化类型Text
 * kout根据需求，输出键指的是单词，类型是String，对应序列化类型是Text
 * vout根据需求，输出值指的是单词的个数，1，类型是int，对应序列化类型是IntWritable
 *
 */
public class WordCountMap extends Mapper<LongWritable, Text, Text, IntWritable> {

    /**
     * 处理分片split中的每一行的数据；针对每行数据，会调用一次map方法
     * 在一次map方法调用时，从一行数据中，获得一个个单词word，再将每个单词word变成键值对形式(word, 1)输出出去
     * 输出的值最终写到本地磁盘中
     * @param key 当前所读行行首相对于split分片开头的字节偏移量
     * @param value  当前所读行
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        //当前行的示例数据(单词间空格分割)：Dear Bear River
        //取得当前行的数据
        String line = value.toString();
        //按照\t进行分割，得到当前行所有单词
        String[] words = line.split("\t");

        for (String word : words) {
            //将每个单词word变成键值对形式(word, 1)输出出去
            //同样，输出前，要将kout, vout包装成对应的可序列化类型，如String对应Text，int对应IntWritable
            context.write(new Text(word), new IntWritable(1));
        }
    }
}

2.4.2 Reducer代码

package com.kaikeba.hadoop.wordcount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 *
 * Reducer<Text, IntWritable, Text, IntWritable>的四个泛型分别表示
 * reduce方法的输入的键的类型kin、输入值的类型vin；输出的键的类型kout、输出的值的类型vout
 * 注意：因为map的输出作为reduce的输入，所以此处的kin、vin类型分别与map的输出的键类型、值类型相同
 * kout根据需求，输出键指的是单词，类型是String，对应序列化类型是Text
 * vout根据需求，输出值指的是每个单词的总个数，类型是int，对应序列化类型是IntWritable
 *
 */
public class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
    /**
     *
     * key相同的一组kv对，会调用一次reduce方法
     * 如reduce task汇聚了众多的键值对，有key是hello的键值对，也有key是spark的键值对，如下
     * (hello, 1)
     * (hello, 1)
     * (hello, 1)
     * (hello, 1)
     * ...
     * (spark, 1)
     * (spark, 1)
     * (spark, 1)
     *
     * 其中，key是hello的键值对被分成一组；merge成[hello, Iterable(1,1,1,1)]，调用一次reduce方法
     * 同样，key是spark的键值对被分成一组；merge成[spark, Iterable(1,1,1)]，再调用一次reduce方法
     *
     * @param key 当前组的key
     * @param values 当前组中，所有value组成的可迭代集和
     * @param context reduce上下文环境对象
     * @throws IOException
     * @throws InterruptedException
     */
    public void reduce(Text key, Iterable<IntWritable> values,
                          Context context) throws IOException, InterruptedException {
        //定义变量，用于累计当前单词出现的次数
        int sum = 0;

        for (IntWritable count : values) {
            //从count中获得值，累加到sum中
            sum += count.get();
        }

        //将单词、单词次数，分别作为键值对，输出
        context.write(key, new IntWritable(sum));// 输出最终结果
    };
}

2.4.3 Main程序入口

package com.kaikeba.hadoop.wordcount;

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;
import java.io.IOException;

/**
 *
 * MapReduce程序入口
 * 注意：
 *  导包时，不要导错了；
 *  另外，map\reduce相关的类，使用mapreduce包下的，是新API，如org.apache.hadoop.mapreduce.Job;；
 */
public class WordCountMain {
    //若在IDEA中本地执行MR程序，需要将mapred-site.xml中的mapreduce.framework.name值修改成local
    //参数 c:/test/README.txt c:/test/wc
    public static void main(String[] args) throws IOException,
            ClassNotFoundException, InterruptedException {

        //判断一下，输入参数是否是两个，分别表示输入路径、输出路径
       if (args.length != 2 || args == null) {
            System.out.println("please input Path!");
            System.exit(0);
        }

        Configuration configuration = new Configuration();
        //configuration.set("mapreduce.framework.name","local");


        //告诉程序，要运行的jar包在哪
        //configuration.set("mapreduce.job.jar","/home/hadoop/IdeaProjects/Hadoop/target/com.kaikeba.hadoop-1.0-SNAPSHOT.jar");

        //调用getInstance方法，生成job实例
        Job job = Job.getInstance(configuration, WordCountMain.class.getSimpleName());

        //设置job的jar包，如果参数指定的类包含在一个jar包中，则此jar包作为job的jar包； 参数class跟主类在一个工程即可；一般设置成主类
//        job.setJarByClass(WordCountMain.class);
        job.setJarByClass(WordCountMain.class);

        //通过job设置输入/输出格式
        //MR的默认输入格式是TextInputFormat，输出格式是TextOutputFormat；所以下两行可以注释掉
//        job.setInputFormatClass(TextInputFormat.class);
//        job.setOutputFormatClass(TextOutputFormat.class);

        //设置输入/输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //设置处理Map阶段的自定义的类
        job.setMapperClass(WordCountMap.class);
        //设置map combine类，减少网路传出量
        job.setCombinerClass(WordCountReduce.class);
        //设置处理Reduce阶段的自定义的类
        job.setReducerClass(WordCountReduce.class);
        //注意：如果map、reduce的输出的kv对类型一致，直接设置reduce的输出的kv对就行；如果不一样，需要分别设置map, reduce的输出的kv类型
        //注意：此处设置的map输出的key/value类型，一定要与自定义map类输出的kv对类型一致；否则程序运行报错
//        job.setMapOutputKeyClass(Text.class);
//        job.setMapOutputValueClass(IntWritable.class);

        //设置reduce task最终输出key/value的类型
        //注意：此处设置的reduce输出的key/value类型，一定要与自定义reduce类输出的kv对类型一致；否则程序运行报错
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 提交作业
        job.waitForCompletion(true);

    }
}

程序运行有两种方式，分别是windows本地运行、集群运行，依次演示

2.5 本地运行

在windows的IDEA中运行

2.5.1 初次运行WordCountMain，先设置main方法参数，根据图示操作即可

• 弹出窗口中，设置包含main方法的类

• 设置main方法在参数

  注意：两个参数间有一个英文空格，表示两个参数

  

  c:/test/README.txt c:/test/wc

2.5.3 本地运行程序

• 在WordCountMain代码上，点击鼠标右键，运行程序
  

2.5.4 查看结果

①成功标识文件

②结果文件

2.6 集群运行

• 用maven插件打jar包；①点击Maven，②双击package打包

• 控制台打印结果；①表示打包成功；②是生成的jar所在路径
  

• 将jar包上传到node01用户主目录/home/hadoop下

• 用hadoop jar命令运行mr程序

[hadoop@node01 ~]$ cd
[hadoop@node01 ~]$ hadoop jar com.kaikeba.hadoop-1.0-SNAPSHOT.jar com.kaikeba.hadoop.wordcount.WordCountMain /README.txt /wordcount01

说明：

com.kaikeba.hadoop-1.0-SNAPSHOT.jar是jar包名

com.kaikeba.hadoop.wordcount.WordCountMain是包含main方法的类的全限定名

/NOTICE.txt和/wordcount是main方法的两个参数，表示输入路径、输出路径

• 确认结果

[hadoop@node01 ~]$ hadoop fs -ls /wordcount01

2.7 总结

• MR分为两个阶段：map阶段、reduce阶段
• MR输入的文件有几个block，就会生成几个map任务
• MR的reduce任务的个数，由程序中编程指定：job.setNumReduceTasks(4)
• map任务
  • map任务中map()一次读取block的一行数据，以kv对的形式输入map()
  • map()的输出作为reduce()的输入
• reduce任务
  • reduce任务通过网络将各执行完成的map任务输出结果中，属于自己的数据取过来
  • key相同的键值对作为一组，调用一次reduce()
  • reduce任务生成一个结果文件
  • 文件写入HDFS

3. WEB UI查看结果

3.1 Yarn

node01是resourcemanager所在节点主机名，根据自己的实际情况修改主机名

浏览器访问url地址：http://node01:8088

3.2 HDFS结果

浏览器输入URL：http://node01:50070

①点击下拉框；②浏览文件系统；③输入根目录，查看hdfs根路径中的内容

4. MapReduce编程（了解一下就ok）：（海量）数据清洗

mapreduce在企业中，可以用于对海量数据的数据清洗；当然，随着新一代大数据框架的出现，也可以使用spark、flink等框架，做数据清洗

4.1 需求

• 现有一批日志文件，日志来源于用户使用搜狗搜索引擎搜索新闻，并点击查看搜索结果过程；
• 但是，日志中有一些记录损坏，现需要使用MapReduce来将这些损坏记录（如记录中少字段、多字段）从日志文件中删除，此过程就是传说中的数据清洗。
• 并且在清洗时，要统计损坏的记录数。

4.2 数据结构

• 日志格式：每行记录有6个字段；分别表示时间datetime、用户ID userid、新闻搜索关键字searchkwd、当前记录在返回列表中的序号retorder、用户点击链接的顺序cliorder、点击的URL连接cliurl

  

  关于retorder、cliorder说明：

  

4.3 逻辑分析

• MapReduce程序一般分为map阶段，将任务分而治之；

• reduce阶段，将map阶段的结果进行聚合；

• 而有些mapreduce应用不需要数据聚合的操作，也就是说不需要reduce阶段。即编程时，不需要编写自定义的reducer类；在main()中调用job.setNumReduceTasks(0)设置

• 而本例的数据清洗就是属于此种情况

• 统计损坏的记录数，可使用自定义计数器的方式进行

  

• map方法的逻辑：取得每一行数据，与每条记录的固定格式比对，是否符合；

  • 若符合，则是完好的记录；
  • 否则是损坏的记录。并对自定义计数器累加

4.4 MR代码

若要集群运行，需先将sogou.50w.utf8上传到HDFS根目录

[hadoop@node01 soft]$ pwd
/kkb/soft
[hadoop@node01 soft]$ hadoop fs -put sogou.50w.utf8 /

运行等操作，与上边类似；不再赘述

4.4.1 Mapper类

具体逻辑，可详见代码注释

注意：实际工作中，写良好的代码注释也是基本的职业素养

package com.kaikeba.hadoop.dataclean;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 *
 * 现对sogou日志数据，做数据清洗；将不符合格式要求的数据删除
 * 每行记录有6个字段；
 * 分别表示时间datetime、用户ID userid、新闻搜索关键字searchkwd、当前记录在返回列表中的序号retorder、用户点击链接的顺序cliorder、点击的URL连接cliurl
 * 日志样本：
 * 20111230111308  0bf5778fc7ba35e657ee88b25984c6e9        nba直播 4       1       http://www.hoopchina.com/tv
 *
 */
public class DataClean {
    /**
     *
     * 基本上大部分MR程序的main方法逻辑，大同小异；将其他MR程序的main方法代码拷贝过来，稍做修改即可
     * 实际开发中，也会有很多的复制、粘贴、修改
     *
     * 注意：若要IDEA中，本地运行MR程序，需要将resources/mapred-site.xml中的mapreduce.framework.name属性值，设置成local
     * @param args
     */
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        //判断一下，输入参数是否是两个，分别表示输入路径、输出路径
        if (args.length != 2 || args == null) {
            System.out.println("please input Path!");
            System.exit(0);
        }

        Configuration configuration = new Configuration();

        //调用getInstance方法，生成job实例
        Job job = Job.getInstance(configuration, DataClean.class.getSimpleName());

        //设置jar包，参数是包含main方法的类
        job.setJarByClass(DataClean.class);

        //设置输入/输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //设置处理Map阶段的自定义的类
        job.setMapperClass(DataCleanMapper.class);

        //注意：此处设置的map输出的key/value类型，一定要与自定义map类输出的kv对类型一致；否则程序运行报错
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        //注意：因为不需要reduce聚合阶段，所以，需要显示设置reduce task个数是0
        job.setNumReduceTasks(0);

        // 提交作业
        job.waitForCompletion(true);
    }

    /**
     * 
     * 自定义mapper类
     * 注意：若自定义的mapper类，与main方法在同一个类中，需要将自定义mapper类，声明成static的
     */
    public static class DataCleanMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
        //为了提高程序的效率，避免创建大量短周期的对象，出发频繁GC；此处生成一个对象，共用
        NullWritable nullValue = NullWritable.get();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //自定义计数器，用于记录残缺记录数
            Counter counter = context.getCounter("DataCleaning", "damagedRecord");

            //获得当前行数据
            //样例数据：20111230111645  169796ae819ae8b32668662bb99b6c2d        塘承高速公路规划线路图  1       1       http://auto.ifeng.com/roll/20111212/729164.shtml
            String line = value.toString();

            //将行数据按照记录中，字段分隔符切分
            String[] fields = line.split("\t");

            //判断字段数组长度，是否为6
            if(fields.length != 6) {
                //若不是，则不输出，并递增自定义计数器
                counter.increment(1L);
            } else {
                //若是6，则原样输出
                context.write(value, nullValue);
            }
        }
    }
}

4.4.2 运行结果

仅以本地运行演示

• 运行

• ①reduce 0%，job就已经successfully，表示此MR程序没有reduce阶段

• ②DataCleaning是自定义计数器组名；damagedRecord是自定义的计数器；值为6，表示有6条损坏记录
  

• 图中part-m-00000中的m表示，此文件是由map任务生成
  

4.5 总结

• MR可用于数据清洗；另外，也可以使用Spark、Flink等组件做数据清洗

• 可使用自定义计数器记录符合特定条件的记录数，用于统计

5. MapReduce编程：用户搜索次数（了解一下就ok）

5.1 需求

• 使用MR编程，统计sogou日志数据中，每个用户搜索的次数；结果写入HDFS

5.2 数据结构

• 数据来源自“MapReduce编程：数据清洗”中的输出结果

• 仍然是sogou日志数据；不再赘述

5.3 逻辑分析

• 还记得之前提到的MR中key的作用吗？
• MR编程时，若要针对某个值对数据进行分组、聚合时，如当前的词频统计例子，需要将每个单词作为reduce输入的key，从而按照单词分组，进而求出每组即每个单词的总次数
• 那么此例也是类似的。
  • 统计每个用户的搜索次数，将userid放到reduce输入的key的位置；
  • 对userid进行分组
  • 进而统计每个用户的搜索次数

5.4 MR代码

给MR程序在IDEA中设置参数，运行等操作，与上边类似；不再赘述

此处MR程序的输入文件是“MapReduce编程：数据清洗”中的输出结果文件

package com.kaikeba.hadoop.searchcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 *
 * 本MR示例，用于统计每个用户搜索并查看URL链接的次数
 */
public class UserSearchCount {

    /**
     *
     * @param args C:\test\datacleanresult c:\test\usersearch
     * @throws IOException
     * @throws ClassNotFoundException
     * @throws InterruptedException
     */
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        //判断一下，输入参数是否是两个，分别表示输入路径、输出路径
        if (args.length != 2 || args == null) {
            System.out.println("please input Path!");
            System.exit(0);
        }

        Configuration configuration = new Configuration();
        //configuration.set("mapreduce.job.jar","/home/hadoop/IdeaProjects/Hadoop/target/com.kaikeba.hadoop-1.0-SNAPSHOT.jar");

        //调用getInstance方法，生成job实例
        Job job = Job.getInstance(configuration, UserSearchCount.class.getSimpleName());

        //设置jar包，参数是包含main方法的类
        job.setJarByClass(UserSearchCount.class);

        //通过job设置输入/输出格式
        //MR的默认输入格式是TextInputFormat，输出格式是TextOutputFormat；所以下两行可以注释掉
//        job.setInputFormatClass(TextInputFormat.class);
//        job.setOutputFormatClass(TextOutputFormat.class);

        //设置输入/输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //设置处理Map阶段的自定义的类
        job.setMapperClass(SearchCountMapper.class);
        //设置map combine类，减少网路传出量
        //job.setCombinerClass(WordCountReduce.class);
        //设置处理Reduce阶段的自定义的类
        job.setReducerClass(SearchCountReducer.class);

        //如果map、reduce的输出的kv对类型一致，直接设置reduce的输出的kv对就行；如果不一样，需要分别设置map, reduce的输出的kv类型
        //注意：此处设置的map输出的key/value类型，一定要与自定义map类输出的kv对类型一致；否则程序运行报错
//        job.setMapOutputKeyClass(Text.class);
//        job.setMapOutputValueClass(IntWritable.class);

        //设置reduce task最终输出key/value的类型
        //注意：此处设置的reduce输出的key/value类型，一定要与自定义reduce类输出的kv对类型一致；否则程序运行报错
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //提交作业
        job.waitForCompletion(true);
    }

    public static class SearchCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        //定义共用的对象，减少GC压力
        Text userIdKOut = new Text();
        IntWritable vOut = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            //获得当前行的数据
            //样例数据：20111230111645  169796ae819ae8b32668662bb99b6c2d        塘承高速公路规划线路图  1       1       http://auto.ifeng.com/roll/20111212/729164.shtml
            String line = value.toString();

            //切分，获得各字段组成的数组
            String[] fields = line.split("\t");

            //因为要统计每个user搜索并查看URL的次数，所以将userid放到输出key的位置
            //注意：MR编程中，根据业务需求设计key是很重要的能力
            String userid = fields[1];

            //设置输出的key的值
            userIdKOut.set(userid);
            //输出结果
            context.write(userIdKOut, vOut);
        }
    }

    public static class SearchCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

        //定义共用的对象，减少GC压力
        IntWritable totalNumVOut = new IntWritable();

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;

            for(IntWritable value: values) {
                sum += value.get();
            }

            //设置当前user搜索并查看总次数
            totalNumVOut.set(sum);
            context.write(key, totalNumVOut);
        }
    }
}

5.4.1 结果

• 运行参数

C:\test\datacleanresult c:\test\usersearch

• 运行结果

5.5 总结

• 结合本例子的需求，设计MR程序；因为要统计每个用户的搜索次数，所以最终userid作为reduce的输出的key
• MR编程能够根据业务需求设计合适的key是一个很重要的能力；而这是需要建立在自己地MR框架原理有清晰认识的基础之上的

6. Shuffle（重点 ）！！！洗牌

• shuffle主要指的是map端的输出作为reduce端输入的过程

6.1 shuffle简图

6.2 shuffle细节图

• 分区用到了分区器，默认分区器是HashPartitioner

  源码：

  public class HashPartitioner<K2, V2> implements Partitioner<K2, V2> {
  
    public void configure(JobConf job) {}
  
    /** Use {@link Object#hashCode()} to partition. */
    public int getPartition(K2 key, V2 value,
                            int numReduceTasks) {
      return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }
  
  }
  

6.3 map端

• 每个map任务都有一个对应的环形内存缓冲区；输出是kv对，先写入到环形缓冲区（默认大小100M），当内容占据80%缓冲区空间后，由一个后台线程将缓冲区中的数据溢出写到一个磁盘文件
• 在溢出写的过程中，map任务可以继续向环形缓冲区写入数据；但是若写入速度大于溢出写的速度，最终造成100m占满后，map任务会暂停向环形缓冲区中写数据的过程；只执行溢出写的过程；直到环形缓冲区的数据全部溢出写到磁盘，才恢复向缓冲区写入
• 后台线程溢写磁盘过程，有以下几个步骤：
  • 先对每个溢写的kv对做分区；分区的个数由MR程序的reduce任务数决定；默认使用HashPartitioner计算当前kv对属于哪个分区；计算公式：(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks
  • 每个分区中，根据kv对的key做内存中排序；
  • 若设置了map端本地聚合combiner，则对每个分区中，排好序的数据做combine操作；
  • 若设置了对map输出压缩的功能，会对溢写数据压缩
• 随着不断的向环形缓冲区中写入数据，会多次触发溢写（每当环形缓冲区写满100m），本地磁盘最终会生成多个溢出文件
• 合并溢写文件：在map task完成之前，所有溢出文件会被合并成一个大的溢出文件；且是已分区、已排序的输出文件
• 小细节：
  • 在合并溢写文件时，如果至少有3个溢写文件，并且设置了map端combine的话，会在合并的过程中触发combine操作；
  • 但是若只有2个或1个溢写文件，则不触发combine操作（因为combine操作，本质上是一个reduce，需要启动JVM虚拟机，有一定的开销）

6.4 reduce端

• reduce task会在每个map task运行完成后，通过HTTP获得map task输出中，属于自己的分区数据（许多kv对）

• 如果map输出数据比较小，先保存在reduce的jvm内存中，否则直接写入reduce磁盘

• 一旦内存缓冲区达到阈值（默认0.66）或map输出数的阈值（默认1000），则触发归并merge，结果写到本地磁盘

• 若MR编程指定了combine，在归并过程中会执行combine操作

• 随着溢出写的文件的增多，后台线程会将它们合并大的、排好序的文件

• reduce task将所有map task复制完后，将合并磁盘上所有的溢出文件

• 默认一次合并10个

• 最后一批合并，部分数据来自内存，部分来自磁盘上的文件

• 进入“归并、排序、分组阶段”

• 每组数据调用一次reduce方法

6.5 总结

• map端
  • map()输出结果先写入环形缓冲区
  • 缓冲区100M；写满80M后，开始溢出写磁盘文件
  • 此过程中，会进行分区、排序、combine（可选）、压缩（可选）
  • map任务完成前，会将多个小的溢出文件，合并成一个大的溢出文件（已分区、排序）
• reduce端
  • 拷贝阶段：reduce任务通过http将map任务属于自己的分区数据拉取过来
  • 开始merge及溢出写磁盘文件
  • 所有map任务的分区全部拷贝过来后，进行阶段合并、排序、分组阶段
  • 每组数据调用一次reduce()
  • 结果写入HDFS

7. 自定义分区（重点）！！！

7.1 分区原理

• 根据之前讲的shuffle，我们知道在map任务中，从环形缓冲区溢出写磁盘时，会先对kv对数据进行分区操作

• 分区操作是由MR中的分区器负责的

• MapReduce有自带的默认分区器

  • HashPartitioner
  • 关键方法getPartition返回当前键值对的分区索引(partition index)

  public class HashPartitioner<K2, V2> implements Partitioner<K2, V2> {
  
    public void configure(JobConf job) {}
  
    /** Use {@link Object#hashCode()} to partition. */
    public int getPartition(K2 key, V2 value, int numReduceTasks) {
      return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }
  }
  

• 环形缓冲区溢出写磁盘前，将每个kv对，作为getPartition()的参数传入；

• 先对键值对中的key求hash值（int类型），与MAX_VALUE按位与；再模上reduce task个数，假设reduce task个数设置为4（可在程序中使用job.setNumReduceTasks(4)指定reduce task个数为4）

  • 那么map任务溢出文件有4个分区，分区index分别是0、1、2、3
  • getPartition()结果有四种：0、1、2、3
  • 根据计算结果，决定当前kv对，落入哪个分区，如结果是0，则当前kv对落入溢出文件的0分区中
  • 最终被相应的reduce task通过http获得

• 若是MR默认分区器，不满足需求；可根据业务逻辑，设计自定义分区器，比如实现图上的功能

7.2 默认分区

程序执行略

代码详见工程com.kaikeba.hadoop.partitioner包

• MR读取三个文件part1.txt、part2.txt、part3.txt；三个文件放到HDFS目录：/customParttitioner中

  

• part1.txt内容如下：

  Dear Bear River
  Dear Car
  

• part2.txt内容如下：

  Car Car River
  Dear Bear
  

• part3.txt内容如下：

  Dear Car Bear
  Car Car
  

• 默认HashPartitioner分区时，查看结果（看代码）

• 运行参数：

/customParttitioner /cp01

• 打jar包运行，结果如下：
  

只有part-r-00001、part-r-00003有数据；另外两个没有数据

HashPartitioner将Bear分到index=1的分区；将Car|Dear|River分到index=3分区

7.3 自定义分区

7.3.1 需求

• 自定义分区，使得文件中，分别以Dear、Bear、River、Car为键的键值对，分别落到index是0、1、2、3的分区中

7.3.2 逻辑分析

• 若要实现以上的分区策略，需要自定义分区类
  • 此类实现Partitioner接口
  • 在getPartition()中实现分区逻辑
• main方法中
  • 设定reduce个数为4
  • 设置自定义的分区类，调用job.setPartitionerClass方法

7.3.3 MR代码

完整代码见代码工程

• 自定义分区类如下

package com.kaikeba.hadoop.partitioner;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

import java.util.HashMap;

public class CustomPartitioner extends Partitioner<Text, IntWritable> {
    public static HashMap<String, Integer> dict = new HashMap<String, Integer>();

    //定义每个键对应的分区index，使用map数据结构完成
    static{
        dict.put("Dear", 0);
        dict.put("Bear", 1);
        dict.put("River", 2);
        dict.put("Car", 3);
    }

    public int getPartition(Text text, IntWritable intWritable, int i) {
        //
        int partitionIndex = dict.get(text.toString());
        return partitionIndex;
    }
}

• 运行结果

结果满足需求

7.4 总结

• 如果默认分区器不满足业务需求，可以自定义分区器
  • 自定义分区器的类继承Partitioner类
  • 覆写getPartition()，在方法中，定义自己的分区策略
  • 在main()方法中调用job.setPartitionerClass()
  • main()中设置reduce任务数

8. 自定义Combiner（重点）！！！本质是reduce,Map端聚合

8.1 需求

• 普通的MR是reduce通过http，取得map任务的分区结果；具体的聚合出结果是在reduce端进行的；

• 以单词计数为例：

  • 下图中的第一个map任务(map1)，本地磁盘中的结果有5个键值对：(Dear, 1)、(Bear, 1)、(River, 1)、(Dear, 1)、(Car, 1)
  • 其中，map1中的两个相同的键值对(Dear, 1)、(Dear, 1)，会被第一个reduce任务(reduce1)通过网络拉取到reduce1端
  • 那么假设map1中(Dear, 1)有1亿个呢？按原思路，map1端需要存储1亿个(Dear, 1)，再将1亿个(Dear, 1)通过网络被reduce1获得，然后再在reduce1端汇总
  • 这样做map端本地磁盘IO、数据从map端到reduce端传输的网络IO比较大
  • 那么想，能不能在reduce1从map1拉取1亿个(Dear, 1)之前，在map端就提前先做下reduce汇总，得到结果(Dear, 100000000)，然后再将这个结果（一个键值对）传输到reduce1呢？
  • 答案是可以的
  • 我们称之为combine操作

• map端combine本地聚合（本质是reduce）

  

8.2 逻辑分析

• 注意：

  • 不论运行多少次Combine操作，都不能影响最终的结果

  • 并非所有的mr都适合combine操作，比如求平均值

    参考：《并非所有MR都适合combine.txt》

• 原理图

  看原图
  

• 当每个map任务的环形缓冲区添满80%，开始溢写磁盘文件

• 此过程会分区、每个分区内按键排序、再combine操作（若设置了combine的话）、若设置map输出压缩的话则再压缩

  • 在合并溢写文件时，如果至少有3个溢写文件，并且设置了map端combine的话，会在合并的过程中触发combine操作；
  • 但是若只有2个或1个溢写文件，则不触发combine操作（因为combine操作，本质上是一个reduce，需要启动JVM虚拟机，有一定的开销）

• combine本质上也是reduce；因为自定义的combine类继承自Reducer父类

• map: (K1, V1) -> list(K2, V2)

• combiner: (K2, list(V2)) -> (K2, V2)

• reduce: (K2, list(V2)) -> (K3, V3)

  • reduce函数与combine函数通常是一样的
  • K3与K2类型相同；
  • V3与V2类型相同
  • 即reduce的输入的kv类型分别与输出的kv类型相同

8.3 MR代码

对原词频统计代码做修改；

详细代码见代码工程

• WordCountMap、WordCountReduce代码保持不变
• 唯一需要做的修改是在WordCountMain中，增加job.setCombinerClass(WordCountReduce.class);
• 修改如下：

8.4 小结

• 使用combine时，首先考虑当前MR是否适合combine
• 总原则是不论使不使用combine不能影响最终的结果
• 在MR时，发生数据倾斜，且可以使用combine时，可以使用combine缓解数据倾斜

9. MR压缩

9.1 需求

• 作用：在MR中，为了减少磁盘IO及网络IO，可考虑在map端、reduce端设置压缩功能
• 给“MapReduce编程：用户搜索次数”代码，增加压缩功能

9.2 逻辑分析

• 那么如何设置压缩功能呢？只需在main方法中，给Configuration对象增加如下设置即可

//开启map输出进行压缩的功能
configuration.set("mapreduce.map.output.compress", "true");
//设置map输出的压缩算法是：BZip2Codec，它是hadoop默认支持的压缩算法，且支持切分
configuration.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.BZip2Codec");
//开启job输出压缩功能
configuration.set("mapreduce.output.fileoutputformat.compress", "true");
//指定job输出使用的压缩算法
configuration.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.BZip2Codec");

9.3 MR代码

• 给“MapReduce编程：用户搜索次数”代码，增加压缩功能，代码如下

  如何打jar包，已演示过，此处不再赘述

package com.kaikeba.hadoop.mrcompress;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 本MR示例，用于统计每个用户搜索并查看URL链接的次数
 */
public class UserSearchCount {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //判断以下，输入参数是否是两个，分别表示输入路径、输出路径
        if (args.length != 2 || args == null) {
            System.out.println("please input Path!");
            System.exit(0);
        }

        Configuration configuration = new Configuration();
        //configuration.set("mapreduce.job.jar","/home/hadoop/IdeaProjects/Hadoop/target/com.kaikeba.hadoop-1.0-SNAPSHOT.jar");
        //开启map输出进行压缩的功能
        configuration.set("mapreduce.map.output.compress", "true");
        //设置map输出的压缩算法是：BZip2Codec，它是hadoop默认支持的压缩算法，且支持切分
        configuration.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.BZip2Codec");
        //开启job输出压缩功能
        configuration.set("mapreduce.output.fileoutputformat.compress", "true");
        //指定job输出使用的压缩算法
        configuration.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.BZip2Codec");

        //调用getInstance方法，生成job实例
        Job job = Job.getInstance(configuration, UserSearchCount.class.getSimpleName());

        //设置jar包，参数是包含main方法的类
        job.setJarByClass(UserSearchCount.class);

        //通过job设置输入/输出格式
        //MR的默认输入格式是TextInputFormat，所以下两行可以注释掉
//        job.setInputFormatClass(TextInputFormat.class);
//        job.setOutputFormatClass(TextOutputFormat.class);

        //设置输入/输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

//        FileOutputFormat.setCompressOutput(job, true);
//        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);

        //设置处理Map阶段的自定义的类
        job.setMapperClass(SearchCountMapper.class);
        //设置map combine类，减少网路传出量
        //job.setCombinerClass(WordCountReduce.class);
        //设置处理Reduce阶段的自定义的类
        job.setReducerClass(SearchCountReducer.class);

        //如果map、reduce的输出的kv对类型一致，直接设置reduce的输出的kv对就行；如果不一样，需要分别设置map, reduce的输出的kv类型
        //注意：此处设置的map输出的key/value类型，一定要与自定义map类输出的kv对类型一致；否则程序运行报错
//        job.setMapOutputKeyClass(Text.class);
//        job.setMapOutputValueClass(IntWritable.class);

        //设置reduce task最终输出key/value的类型
        //注意：此处设置的reduce输出的key/value类型，一定要与自定义reduce类输出的kv对类型一致；否则程序运行报错
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 提交作业
        job.waitForCompletion(true);
    }

    public static class SearchCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        //定义共用的对象，减少GC压力
        Text userIdKOut = new Text();
        IntWritable vOut = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //获得当前行的数据
            //样例数据：20111230111645  169796ae819ae8b32668662bb99b6c2d        塘承高速公路规划线路图  1       1       http://auto.ifeng.com/roll/20111212/729164.shtml
            String line = value.toString();

            //切分，获得各字段组成的数组
            String[] fields = line.split("\t");

            //因为要统计每个user搜索并查看URL的次数，所以将userid放到输出key的位置
            //注意：MR编程中，根据业务需求设计key是很重要的能力
            String userid = fields[1];

            //设置输出的key的值
            userIdKOut.set(userid);
            //输出结果
            context.write(userIdKOut, vOut);
        }
    }

    public static class SearchCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        //定义共用的对象，减少GC压力
        IntWritable totalNumVOut = new IntWritable();

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;

            for(IntWritable value: values) {
                sum += value.get();
            }

            //设置当前user搜索并查看总次数
            totalNumVOut.set(sum);
            context.write(key, totalNumVOut);
        }
    }
}

• 生成jar包，并运行jar包

[hadoop@node01 target]$ hadoop jar com.kaikeba.hadoop-1.0-SNAPSHOT.jar com.kaikeba.hadoop.mrcompress.UserSearchCount /sogou.2w.utf8 /compressed

• 查看结果

  可增加数据量，查看使用压缩算法前后的系统各计数器的数据量变化

[hadoop@node01 target]$ hadoop fs -ls -h /compressed

9.4 总结

• MR过程中使用压缩可减少数据量，进而减少磁盘IO、网络IO数据量
• 可设置map端输出的压缩
• 可设置job最终结果的压缩
• 通过相应的配置项即可实现

10. 自定义InputFormat（难点）！！！！！

10.1 MapReduce执行过程

• 上图也描述了mapreduce的一个完整的过程；我们主要看map任务是如何从hdfs读取分片数据的部分

  • 涉及3个关键的类

  • ①InputFormat输入格式类

    ②InputSplit输入分片类：getSplits()

    • InputFormat输入格式类将输入文件分成一个个分片InputSplit
    • 每个Map任务对应一个split分片

    ③RecordReader记录读取器类：createRecordReader()

    • RecordReader（记录读取器）读取分片数据，一行记录生成一个键值对
    • 传入map任务的map()方法，调用map()

• 所以，如果需要根据自己的业务情况，自定义输入的话，需要自定义两个类：

  • InputFormat类
  • RecordReader类

• 详细流程：

  • 客户端调用InputFormat的**getSplits()**方法，获得输入文件的分片信息

    

  • 针对每个MR job会生成一个相应的app master，负责map\reduce任务的调度及监控执行情况

  • 将分片信息传递给MR job的app master

  • app master根据分片信息，尽量将map任务尽量调度在split分片数据所在节点（移动计算不移动数据）

    

  • 有几个分片，就生成几个map任务

  • 每个map任务将split分片传递给createRecordReader()方法，生成此分片对应的RecordReader

  • RecordReader用来读取分片的数据，生成记录的键值对

    • nextKeyValue()判断是否有下一个键值对，如果有，返回true；否则，返回false
    • 如果返回true，调用getCurrentKey()获得当前的键
    • 调用getCurrentValue()获得当前的值

  • map任务运行过程

    

    • map任务运行时，会调用run()

    • 首先运行一次setup()方法；只在map任务启动时，运行一次；一些初始化的工作可以在setup方法中完成；如要连接数据库之类的操作

    • while循环，调用context.nextKeyValue()；会委托给RecordRecord的nextKeyValue()，判断是否有下一个键值对

    • 如果有下一个键值对，调用context.getCurrentKey()、context.getCurrentValue()获得当前的键、值的值（也是调用RecordReader的同名方法）

      

    • 作为参数传入map(key, value, context)，调用一次map()

    • 当读取分片尾，context.nextKeyValue()返回false；退出循环

    • 调用cleanup()方法，只在map任务结束之前，调用一次；所以，一些回收资源的工作可在此方法中实现，如关闭数据库连接

10.2 需求

• 无论hdfs还是mapreduce，处理小文件都有损效率，实践中，又难免面临处理大量小文件的场景，此时，就需要有相应解决方案

10.3 逻辑分析

• 小文件的优化无非以下几种方式：
  • 在数据采集的时候，就将小文件或小批数据合成大文件再上传HDFS(SequenceFile方案)
  • 在业务处理之前，在HDFS上使用mapreduce程序对小文件进行合并；可使用自定义InputFormat实现
  • 在mapreduce处理时，可采用CombineFileInputFormat提高效率
• 本例使用第二种方案，自定义输入格式

10.4 MR代码

• 自定义InputFormat

  package com.kaikeba.hadoop.inputformat;
  
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.BytesWritable;
  import org.apache.hadoop.io.NullWritable;
  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;
  
  /**
   * 自定义InputFormat类；
   * 泛型：
   *  键：因为不需要使用键，所以设置为NullWritable
   *  值：值用于保存小文件的内容，此处使用BytesWritable
   */
  public class WholeFileInputFormat extends FileInputFormat<NullWritable, BytesWritable> {
  
      /**
       *
       * 返回false，表示输入文件不可切割
       * @param context
       * @param file
       * @return
       */
      @Override
      protected boolean isSplitable(JobContext context, Path file) {
          return false;
      }
  
      /**
       * 生成读取分片split的RecordReader
       * @param split
       * @param context
       * @return
       * @throws IOException
       * @throws InterruptedException
       */
      @Override
      public RecordReader<NullWritable, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException,InterruptedException {
          //使用自定义的RecordReader类
          WholeFileRecordReader reader = new WholeFileRecordReader();
          //初始化RecordReader
          reader.initialize(split, context);
          return reader;
      }
  }
  

• 自定义RecordReader

  实现6个相关方法

  package com.kaikeba.hadoop.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.NullWritable;
  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;
  
  /**
   *
   * RecordReader的核心工作逻辑：
   * 通过nextKeyValue()方法去读取数据构造将返回的key   value
   * 通过getCurrentKey 和 getCurrentValue来返回上面构造好的key和value
   *
   * @author
   */
  public class WholeFileRecordReader extends RecordReader<NullWritable, BytesWritable> {
  
      //要读取的分片
      private FileSplit fileSplit;
      private Configuration conf;
  
      //读取的value数据
      private BytesWritable value = new BytesWritable();
      /**
       *
       * 标识变量，分片是否已被读取过；因为小文件设置成了不可切分，所以一个小文件只有一个分片；
       * 而这一个分片的数据，只读取一次，一次读完所有数据
       * 所以设置此标识
       */
      private boolean processed = false;
  
      /**
       * 初始化
       * @param split
       * @param context
       * @throws IOException
       * @throws InterruptedException
       */
      @Override
      public void initialize(InputSplit split, TaskAttemptContext context)
              throws IOException, InterruptedException {
          this.fileSplit = (FileSplit) split;
          this.conf = context.getConfiguration();
      }
  
      /**
       * 判断是否有下一个键值对。若有，则读取分片中的所有的数据
       * @return
       * @throws IOException
       * @throws InterruptedException
       */
      @Override
      public boolean nextKeyValue() throws IOException, InterruptedException {
          if (!processed) {
              byte[] contents = new byte[(int) fileSplit.getLength()];
              Path file = fileSplit.getPath();
              FileSystem fs = file.getFileSystem(conf);
              FSDataInputStream in = null;
              try {
                  in = fs.open(file);
                  IOUtils.readFully(in, contents, 0, contents.length);
                  value.set(contents, 0, contents.length);
              } finally {
                  IOUtils.closeStream(in);
              }
              processed = true;
              return true;
          }
          return false;
      }
  
      /**
       * 获得当前的key
       * @return
       * @throws IOException
       * @throws InterruptedException
       */
      @Override
      public NullWritable getCurrentKey() throws IOException,
              InterruptedException {
          return NullWritable.get();
      }
  
      /**
       * 获得当前的value
       * @return
       * @throws IOException
       * @throws InterruptedException
       */
      @Override
      public BytesWritable getCurrentValue() throws IOException,
              InterruptedException {
          return value;
      }
  
      /**
       * 获得分片读取的百分比；因为如果读取分片数据的话，会一次性的读取完；所以进度要么是1，要么是0
       * @return
       * @throws IOException
       */
      @Override
      public float getProgress() throws IOException {
          //因为一个文件作为一个整体处理，所以，如果processed为true，表示已经处理过了，进度为1；否则为0
          return processed ? 1.0f : 0.0f;
      }
  
      @Override
      public void close() throws IOException {
      }
  }
  

• main方法

  package com.kaikeba.hadoop.inputformat;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.conf.Configured;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.BytesWritable;
  import org.apache.hadoop.io.NullWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapreduce.InputSplit;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;
  
  import java.io.IOException;
  
  /**
   * 让主类继承Configured类，实现Tool接口
   * 实现run()方法
   * 将以前main()方法中的逻辑，放到run()中
   * 在main()中，调用ToolRunner.run()方法，第一个参数是当前对象；第二个参数是输入、输出
   */
  public class SmallFiles2SequenceFile extends Configured implements Tool {
  
      /**
       * 自定义Mapper类
       * mapper类的输入键值对类型，与自定义InputFormat的输入键值对保持一致
       * mapper类的输出的键值对类型，分别是文件名、文件内容
       */
      static class SequenceFileMapper extends
              Mapper<NullWritable, BytesWritable, Text, BytesWritable> {
  
          private Text filenameKey;
  
          /**
           * 取得文件名
           * @param context
           * @throws IOException
           * @throws InterruptedException
           */
          @Override
          protected void setup(Context context) throws IOException,
                  InterruptedException {
              InputSplit split = context.getInputSplit();
              //获得当前文件路径
              Path path = ((FileSplit) split).getPath();
              filenameKey = new Text(path.toString());
          }
  
          @Override
          protected void map(NullWritable key, BytesWritable value,
                             Context context) throws IOException, InterruptedException {
              context.write(filenameKey, value);
          }
      }
  
      public int run(String[] args) throws Exception {
          Configuration conf = new Configuration();
          Job job = Job.getInstance(conf,"combine small files to sequencefile");
          job.setJarByClass(SmallFiles2SequenceFile.class);
  
          //设置自定义输入格式
          job.setInputFormatClass(WholeFileInputFormat.class);
  
          WholeFileInputFormat.addInputPath(job,new Path(args[0]));
          //设置输出格式SequenceFileOutputFormat及输出路径
          job.setOutputFormatClass(SequenceFileOutputFormat.class);
          SequenceFileOutputFormat.setOutputPath(job,new Path(args[1]));
  
          job.setOutputKeyClass(Text.class);
          job.setOutputValueClass(BytesWritable.class);
          job.setMapperClass(SequenceFileMapper.class);
          return job.waitForCompletion(true) ? 0 : 1;
      }
  
      public static void main(String[] args) throws Exception {
          int exitCode = ToolRunner.run(new SmallFiles2SequenceFile(),
                  args);
          System.exit(exitCode);
  
      }
  }
  

10.5 总结

• 若要自定义InputFormat的话
  • 需要自定义InputFormat类，并覆写getRecordReader()方法
  • 自定义RecordReader类，实现方法
    • initialize()
    • nextKeyValue()
    • getCurrentKey()
    • getCurrentValue()
    • getProgress()
    • close()

11. 自定义OutputFormat

11.1 需求

• 现在有一些订单的评论数据，要将订单的好评与其它级别的评论（中评、差评）进行区分开来，将最终的数据分开到不同的文件夹下面去

• 数据第九个字段表示评分等级：0 好评，1 中评，2 差评

  

11.2 逻辑分析

• 程序的关键点是在一个mapreduce程序中，根据数据的不同(好评的评级不同)，输出两类结果到不同目录
• 这类灵活的输出，需求通过自定义OutputFormat来实现

11.3 实现要点

• 在mapreduce中访问外部资源
• 自定义OutputFormat类，覆写getRecordWriter()方法
• 自定义RecordWriter类，覆写具体输出数据的方法write()

11.4 MR代码

• 自定义OutputFormat

package com.kaikeba.hadoop.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;

/**
 *
 * 本例使用框架默认的Reducer，它将Mapper输入的kv对，原样输出；所以reduce输出的kv类型分别是Text, NullWritable
 * 自定义OutputFormat的类，泛型表示reduce输出的键值对类型；要保持一致;
 * map--(kv)-->reduce--(kv)-->OutputFormat
 */
public class MyOutPutFormat extends FileOutputFormat<Text, NullWritable> {

    /**
     * 两个输出文件;
     * good用于保存好评文件；其它评级保存到bad中
     * 根据实际情况修改path;node01及端口号8020
     */
    String bad = "hdfs://node01:8020/outputformat/bad/r.txt";
    String good = "hdfs://node01:8020/outputformat/good/r.txt";

    /**
     *
     * @param context
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        //获得文件系统对象
        FileSystem fs = FileSystem.get(context.getConfiguration());
        //两个输出文件路径
        Path badPath = new Path(bad);
        Path goodPath = new Path(good);
        FSDataOutputStream badOut = fs.create(badPath);
        FSDataOutputStream goodOut = fs.create(goodPath);
        return new MyRecordWriter(badOut,goodOut);
    }

    /**
     * 泛型表示reduce输出的键值对类型；要保持一致
     */
    static class MyRecordWriter extends RecordWriter<Text, NullWritable>{

        FSDataOutputStream badOut = null;
        FSDataOutputStream goodOut = null;

        public MyRecordWriter(FSDataOutputStream badOut, FSDataOutputStream goodOut) {
            this.badOut = badOut;
            this.goodOut = goodOut;
        }

        /**
         * 自定义输出kv对逻辑
         * @param key
         * @param value
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        public void write(Text key, NullWritable value) throws IOException, InterruptedException {
            if (key.toString().split("\t")[9].equals("0")){//好评
                goodOut.write(key.toString().getBytes());
                goodOut.write("\r\n".getBytes());
            }else{//其它评级
                badOut.write(key.toString().getBytes());
                badOut.write("\r\n".getBytes());
            }
        }

        /**
         * 关闭流
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            if(goodOut !=null){
                goodOut.close();
            }
            if(badOut !=null){
                badOut.close();
            }
        }
    }
}

• main方法

package com.kaikeba.hadoop.outputformat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

public class MyOwnOutputFormatMain extends Configured implements Tool {

    public int run(String[] args) throws Exception {
        Configuration conf = super.getConf();
        Job job = Job.getInstance(conf, MyOwnOutputFormatMain.class.getSimpleName());
        job.setJarByClass(MyOwnOutputFormatMain.class);

        //默认项，可以省略或者写出也可以
        //job.setInputFormatClass(TextInputFormat.class);
        //设置输入文件
        TextInputFormat.addInputPath(job, new Path(args[0]));
        job.setMapperClass(MyMapper.class);
        //job.setMapOutputKeyClass(Text.class);
        //job.setMapOutputValueClass(NullWritable.class);

        //设置自定义的输出类
        job.setOutputFormatClass(MyOutPutFormat.class);
        //设置一个输出目录，这个目录会输出一个success的成功标志的文件
        MyOutPutFormat.setOutputPath(job, new Path(args[1]));
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        //默认项，即默认有一个reduce任务，所以可以省略
//        job.setNumReduceTasks(1);
//        //Reducer将输入的键值对原样输出
//        job.setReducerClass(Reducer.class);

        boolean b = job.waitForCompletion(true);
        return b ? 0: 1;
    }

    /**
     *
     * Mapper输出的key、value类型
     * 文件每行的内容作为输出的key，对应Text类型
     * 输出的value为null，对应NullWritable
     */
    public static class MyMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //把当前行内容作为key输出；value为null
            context.write(value, NullWritable.get());
        }
    }

    /**
     *
     * @param args /ordercomment.csv /ofo
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        ToolRunner.run(configuration, new MyOwnOutputFormatMain(), args);
    }
}

11.5 总结

• 自定义outputformat

  • 泛型与reduce输出的键值对类型保持一致
  • 覆写getRecordWriter()方法

• 自定义RecordWriter

  • 泛型与reduce输出的键值对类型保持一致
  • 覆写具体输出数据的方法write()、close()

• main方法

  • job.setOutputFormatClass使用自定义在输出类

12. 二次排序（重点 ）

12.1 需求

• 数据：有一个简单的关于员工工资的记录文件salary.txt

  • 每条记录如下，有3个字段，分别表示name、age、salary

  • nancy 22 8000

• 使用MR处理记录，实现结果中

  • 按照工资从高到低的降序排序
  • 若工资相同，则按年龄升序排序

12.2 逻辑分析

• 利用MR中key具有可比较的特点

• MapReduce中，根据key进行分区、排序、分组

• 有些MR的输出的key可以直接使用hadoop框架的可序列化可比较类型表示，如Text、IntWritable等等，而这些类型本身是可比较的；如IntWritable默认升序排序

  

• 但有时，使用MR编程，输出的key，若使用hadoop自带的key类型无法满足需求

  • 此时，需要自定义的key类型（包含的是非单一信息，如此例包含工资、年龄）；
  • 并且也得是**可序列化、可比较的**

• 需要自定义key，定义排序规则

  • 实现：按照人的salary降序排序，若相同，则再按age升序排序；若salary、age相同，则放入同一组

12.3 MR代码

• 详见工程代码
• 自定义key类型Person类

package com.kaikeba.hadoop.secondarysort;

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

//根据输入文件格式，定义JavaBean，作为MR时，Map的输出key类型；要求此类可序列化、可比较
public class Person implements WritableComparable<Person> {
    private String name;
    private int age;
    private int salary;

    public Person() {
    }

    public Person(String name, int age, int salary) {
        //super();
        this.name = name;
        this.age = age;
        this.salary = salary;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public int getSalary() {
        return salary;
    }

    public void setSalary(int salary) {
        this.salary = salary;
    }

    @Override
    public String toString() {
        return this.salary + "  " + this.age + "    " + this.name;
    }

    //两个Person对象的比较规则：①先比较salary，高的排序在前；②若相同，age小的在前
    public int compareTo(Person other) {
        int compareResult= this.salary - other.salary;
        if(compareResult != 0) {//若两个人工资不同
            //工资降序排序；即工资高的排在前边
            return -compareResult;
        } else {//若工资相同
            //年龄升序排序；即年龄小的排在前边
            return this.age - other.age;
        }
    }

    //序列化，将NewKey转化成使用流传送的二进制
    public void write(DataOutput dataOutput) throws IOException {
        //注意：①使用正确的write方法；②记住此时的序列化的顺序，name、age、salary
        dataOutput.writeUTF(name);
        dataOutput.writeInt(age);
        dataOutput.writeInt(salary);
    }

    //使用in读字段的顺序，要与write方法中写的顺序保持一致：name、age、salary
    public void readFields(DataInput dataInput) throws IOException {
        //read string
        //注意：①使用正确的read方法；②读取顺序与write()中序列化的顺序保持一致
        this.name = dataInput.readUTF();
        this.age = dataInput.readInt();
        this.salary = dataInput.readInt();
    }
}

• main类、mapper、reducer

package com.kaikeba.hadoop.secondarysort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.net.URI;

public class SecondarySort {

	/**
	 *
	 * @param args /salary.txt /secondarysort
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		Configuration configuration = new Configuration();
		//configuration.set("mapreduce.job.jar","/home/bruce/project/kkbhdp01/target/com.kaikeba.hadoop-1.0-SNAPSHOT.jar");
		Job job = Job.getInstance(configuration, SecondarySort.class.getSimpleName());

		FileSystem fileSystem = FileSystem.get(URI.create(args[1]), configuration);
		//生产中慎用
		if (fileSystem.exists(new Path(args[1]))) {
			fileSystem.delete(new Path(args[1]), true);
		}

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		job.setMapperClass(MyMap.class);
		//由于mapper与reducer输出的kv类型分别相同，所以，下两行可以省略
//		job.setMapOutputKeyClass(Person.class);
//		job.setMapOutputValueClass(NullWritable.class);
		
		//设置reduce的个数;默认为1
		//job.setNumReduceTasks(1);

		job.setReducerClass(MyReduce.class);
		job.setOutputKeyClass(Person.class);
		job.setOutputValueClass(NullWritable.class);
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		job.waitForCompletion(true);

	}

	//MyMap的输出key用自定义的Person类型；输出的value为null
	public static class MyMap extends Mapper<LongWritable, Text, Person, NullWritable> {
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {

			String[] fields = value.toString().split("\t");
			String name = fields[0];
			int age = Integer.parseInt(fields[1]);
			int salary = Integer.parseInt(fields[2]);
			//在自定义类中进行比较
			Person person = new Person(name, age, salary);
			//person对象作为输出的key
			context.write(person, NullWritable.get());
		}
	}

	public static class MyReduce extends Reducer<Person, NullWritable, Person, NullWritable> {
		@Override
		protected void reduce(Person key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
			//输入的kv对，原样输出
			context.write(key, NullWritable.get());
		}
	}
}

12.4 总结

• 如果MR时，key的排序规则比较复杂，比如需要根据字段1排序，若字段1相同，则需要根据字段2排序…，此时，可以使用自定义key实现
• 将自定义的key作为MR中，map输出的key的类型（reduce输入的类型）
• 自定义的key
  • 实现WritableComparable接口
  • 实现compareTo比较方法
  • 实现write序列化方法
  • 实现readFields反序列化方法

13. 自定义分组求topN（重难点）

13.1 需求

• 现有一个淘宝用户订单历史记录文件；每条记录有6个字段，分别表示

  • userid、datetime、title商品标题、unitPrice商品单价、purchaseNum购买量、productId商品ID

  

• 现使用MR编程，求出每个用户、每个月消费金额最多的两笔订单，花了多少钱

  • 所以得相同用户、同一个年月的数据，分到同一组

13.2 逻辑分析

• 根据文件格式，自定义JavaBean类OrderBean
  • 实现WritableComparable接口
  • 包含6个字段分别对应文件中的6个字段
  • 重点实现compareTo方法
    • 先比较userid是否相等；若不相等，则userid升序排序
    • 若相等，比较两个Bean的日期是否相等；若不相等，则日期升序排序
    • 若相等，再比较总开销，降序排序
  • 实现序列化方法write()
  • 实现反序列化方法readFields()
• 自定义分区类
  • 继承Partitioner类
  • getPartiton()实现，userid相同的，处于同一个分区
• 自定义Mapper类
  • 输出key是当前记录对应的Bean对象
  • 输出的value对应当前订单的总开销
• 自定义分组类
  • 决定userid相同、日期（年月）相同的记录，分到同一组中，调用一次reduce()
• 自定义Reduce类
  • reduce()中求出当前一组数据中，开销头两笔的信息
• main方法
  • job.setMapperClass
  • job.setPartitionerClass
  • job.setReducerClass
  • job.setGroupingComparatorClass

13.3 MR代码

详细代码见代码工程

• OrderBean

package com.kaikeba.hadoop.grouping;

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

//实现WritableComparable接口
public class OrderBean implements WritableComparable<OrderBean> {

    //用户ID
    private String userid;
    //年月
    //year+month -> 201408
    private String datetime;
    //标题
    private String title;
    //单价
    private double unitPrice;
    //购买量
    private int purchaseNum;
    //商品ID
    private String produceId;

    public OrderBean() {
    }

    public OrderBean(String userid, String datetime, String title, double unitPrice, int purchaseNum, String produceId) {
        super();
        this.userid = userid;
        this.datetime = datetime;
        this.title = title;
        this.unitPrice = unitPrice;
        this.purchaseNum = purchaseNum;
        this.produceId = produceId;
    }

    //key的比较规则
    public int compareTo(OrderBean other) {
        //OrderBean作为MR中的key；如果对象中的userid相同，即ret1为0；就表示两个对象是同一个用户
        int ret1 = this.userid.compareTo(other.userid);

        if (ret1 == 0) {
            //如果userid相同，比较年月
            String thisYearMonth = this.getDatetime();
            String otherYearMonth = other.getDatetime();
            int ret2 = thisYearMonth.compareTo(otherYearMonth);

            if(ret2 == 0) {//若datetime相同
                //如果userid、年月都相同，比较单笔订单的总开销
                Double thisTotalPrice = this.getPurchaseNum()*this.getUnitPrice();
                Double oTotalPrice = other.getPurchaseNum()*other.getUnitPrice();
                //总花销降序排序；即总花销高的排在前边
                return -thisTotalPrice.compareTo(oTotalPrice);
            } else {
                //若datatime不同，按照datetime升序排序
                return ret2;
            }
        } else {
            //按照userid升序排序
            return ret1;
        }
    }

    /**
     * 序列化
     * @param dataOutput
     * @throws IOException
     */
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(userid);
        dataOutput.writeUTF(datetime);
        dataOutput.writeUTF(title);
        dataOutput.writeDouble(unitPrice);
        dataOutput.writeInt(purchaseNum);
        dataOutput.writeUTF(produceId);
    }

    /**
     * 反序列化
     * @param dataInput
     * @throws IOException
     */
    public void readFields(DataInput dataInput) throws IOException {
        this.userid = dataInput.readUTF();
        this.datetime = dataInput.readUTF();
        this.title = dataInput.readUTF();
        this.unitPrice = dataInput.readDouble();
        this.purchaseNum = dataInput.readInt();
        this.produceId = dataInput.readUTF();
    }

    /**
     * 使用默认分区器，那么userid相同的，落入同一分区；
     * 另外一个方案：此处不覆写hashCode方法，而是自定义分区器，getPartition方法中，对OrderBean的userid求hashCode值%reduce任务数
     * @return
     */
//    @Override
//    public int hashCode() {
//        return this.userid.hashCode();
//    }

    @Override
    public String toString() {
        return "OrderBean{" +
                "userid='" + userid + '\'' +
                ", datetime='" + datetime + '\'' +
                ", title='" + title + '\'' +
                ", unitPrice=" + unitPrice +
                ", purchaseNum=" + purchaseNum +
                ", produceId='" + produceId + '\'' +
                '}';
    }

    public String getUserid() {
        return userid;
    }

    public void setUserid(String userid) {
        this.userid = userid;
    }

    public String getDatetime() {
        return datetime;
    }

    public void setDatetime(String datetime) {
        this.datetime = datetime;
    }

    public String getTitle() {
        return title;
    }

    public void setTitle(String title) {
        this.title = title;
    }

    public double getUnitPrice() {
        return unitPrice;
    }

    public void setUnitPrice(double unitPrice) {
        this.unitPrice = unitPrice;
    }

    public int getPurchaseNum() {
        return purchaseNum;
    }

    public void setPurchaseNum(int purchaseNum) {
        this.purchaseNum = purchaseNum;
    }

    public String getProduceId() {
        return produceId;
    }

    public void setProduceId(String produceId) {
        this.produceId = produceId;
    }
}

• MyPartitioner

package com.kaikeba.hadoop.grouping;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.mapreduce.Partitioner;

//mapper的输出key类型是自定义的key类型OrderBean；输出value类型是单笔订单的总开销double -> DoubleWritable
public class MyPartitioner extends Partitioner<OrderBean, DoubleWritable> {
    @Override
    public int getPartition(OrderBean orderBean, DoubleWritable doubleWritable, int numReduceTasks) {
        //userid相同的，落入同一分区
        return (orderBean.getUserid().hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }
}

• MyMapper

package com.kaikeba.hadoop.grouping;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * 输出kv，分别是OrderBean、用户每次下单的总开销
 */
public class MyMapper extends Mapper<LongWritable, Text, OrderBean, DoubleWritable> {
    DoubleWritable valueOut = new DoubleWritable();
    DateUtils dateUtils = new DateUtils();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //13764633023     2014-12-01 02:20:42.000 全视目Allseelook 原宿风暴显色美瞳彩色隐形艺术眼镜1片 拍2包邮    33.6    2       18067781305
        String record = value.toString();
        String[] fields = record.split("\t");
        if(fields.length == 6) {
            String userid = fields[0];
            String datetime = fields[1];
            String yearMonth = dateUtils.getYearMonthString(datetime);
            String title = fields[2];
            double unitPrice = Double.parseDouble(fields[3]);
            int purchaseNum = Integer.parseInt(fields[4]);
            String produceId = fields[5];

            //生成OrderBean对象
            OrderBean orderBean = new OrderBean(userid, yearMonth, title, unitPrice, purchaseNum, produceId);

            //此订单的总开销
            double totalPrice = unitPrice * purchaseNum;
            valueOut.set(totalPrice);

            context.write(orderBean, valueOut);
        }
    }
}

• MyReducer

package com.kaikeba.hadoop.grouping;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

//输出的key为userid拼接上年月的字符串，对应Text；输出的value对应单笔订单的金额
public class MyReducer extends Reducer<OrderBean, DoubleWritable, Text, DoubleWritable> {
    /**
     * ①由于自定义分组逻辑，相同用户、相同年月的订单是一组，调用一次reduce()；
     * ②由于自定义的key类OrderBean中，比较规则compareTo规定，相同用户、相同年月的订单，按总金额降序排序
     * 所以取出头两笔，就实现需求
     * @param key
     * @param values
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(OrderBean key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
        //求每个用户、每个月、消费金额最多的两笔多少钱
        int num = 0;
        for(DoubleWritable value: values) {
            if(num < 2) {
                String keyOut = key.getUserid() + "  " + key.getDatetime();
                context.write(new Text(keyOut), value);
                num++;
            } else {
                break;
            }
        }

    }
}

• MyGroup

package com.kaikeba.hadoop.grouping;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

//自定义分组类：reduce端调用reduce()前，对数据做分组；每组数据调用一次reduce()
public class MyGroup extends WritableComparator {

    public MyGroup() {
        //第一个参数表示key class
        super(OrderBean.class, true);
    }

    //分组逻辑
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        //userid相同，且同一月的分成一组
        OrderBean aOrderBean = (OrderBean)a;
        OrderBean bOrderBean = (OrderBean)b;

        String aUserId = aOrderBean.getUserid();
        String bUserId = bOrderBean.getUserid();

        //userid、年、月相同的，作为一组
        int ret1 = aUserId.compareTo(bUserId);
        if(ret1 == 0) {//同一用户
            //年月也相同返回0，在同一组；
            return aOrderBean.getDatetime().compareTo(bOrderBean.getDatetime());
        } else {
            return ret1;
        }
    }
}

• CustomGroupingMain

package com.kaikeba.hadoop.grouping;

import com.kaikeba.hadoop.wordcount.WordCountMain;
import com.kaikeba.hadoop.wordcount.WordCountMap;
import com.kaikeba.hadoop.wordcount.WordCountReduce;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
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;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

public class CustomGroupingMain extends Configured implements Tool {

    ///tmall-201412-test.csv /cgo
    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new CustomGroupingMain(), args);
        System.exit(exitCode);
    }

    @Override
    public int run(String[] args) throws Exception {
        //判断以下，输入参数是否是两个，分别表示输入路径、输出路径
        if (args.length != 2 || args == null) {
            System.out.println("please input Path!");
            System.exit(0);
        }

        Configuration configuration = new Configuration();
        //告诉程序，要运行的jar包在哪
        //configuration.set("mapreduce.job.jar","/home/hadoop/IdeaProjects/Hadoop/target/com.kaikeba.hadoop-1.0-SNAPSHOT.jar");

        //调用getInstance方法，生成job实例
        Job job = Job.getInstance(configuration, CustomGroupingMain.class.getSimpleName());

        //设置jar包，参数是包含main方法的类
        job.setJarByClass(CustomGroupingMain.class);

        //通过job设置输入/输出格式
        //MR的默认输入格式是TextInputFormat，所以下两行可以注释掉
//        job.setInputFormatClass(TextInputFormat.class);
//        job.setOutputFormatClass(TextOutputFormat.class);

        //设置输入/输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //设置处理Map阶段的自定义的类
        job.setMapperClass(MyMapper.class);
        //设置map combine类，减少网路传出量
        //job.setCombinerClass(MyReducer.class);
        job.setPartitionerClass(MyPartitioner.class);
        //设置处理Reduce阶段的自定义的类
        job.setReducerClass(MyReducer.class);
        job.setGroupingComparatorClass(MyGroup.class);

        //如果map、reduce的输出的kv对类型一致，直接设置reduce的输出的kv对就行；如果不一样，需要分别设置map, reduce的输出的kv类型
        //注意：此处设置的map输出的key/value类型，一定要与自定义map类输出的kv对类型一致；否则程序运行报错
        job.setMapOutputKeyClass(OrderBean.class);
        job.setMapOutputValueClass(DoubleWritable.class);

        //设置reduce task最终输出key/value的类型
        //注意：此处设置的reduce输出的key/value类型，一定要与自定义reduce类输出的kv对类型一致；否则程序运行报错
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        // 提交作业
        return job.waitForCompletion(true) ? 0 : 1;
    }
}

• DateUtils

package com.kaikeba.hadoop.grouping;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

public class DateUtils {

    public static void main(String[] args) {
        //test1
//        String str1 = "13764633024  2014-10-01 02:20:42.000";
//        String str2 = "13764633023  2014-11-01 02:20:42.000";
//        System.out.println(str1.compareTo(str2));

        //test2
//        String datetime = "2014-12-01 02:20:42.000";
//        LocalDateTime localDateTime = parseDateTime(datetime);
//        int year = localDateTime.getYear();
//        int month = localDateTime.getMonthValue();
//        int day = localDateTime.getDayOfMonth();
//        System.out.println("year-> " + year + "; month -> " + month + "; day -> " + day);

        //test3
//        String datetime = "2014-12-01 02:20:42.000";
//        System.out.println(getYearMonthString(datetime));
    }

    public LocalDateTime parseDateTime(String dateTime) {
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
        LocalDateTime localDateTime = LocalDateTime.parse(dateTime, formatter);
        return localDateTime;
    }

    //日期格式转换工具类：将2014-12-14 20:42:14.000转换成201412
    public String getYearMonthString(String dateTime) {
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
        LocalDateTime localDateTime = LocalDateTime.parse(dateTime, formatter);
        int year = localDateTime.getYear();
        int month = localDateTime.getMonthValue();
        return year + "" + month;
    }



}

13.4 总结

• 要实现自定义分组逻辑
  • 一般会自定义JavaBean，作为map输出的key
    • 实现其中的compareTo方法，设置key的比较逻辑
    • 实现序列化方法write()
    • 实现反序列化方法readFields()
  • 自定义mapper类、reducer类
  • 自定义partition类，getPartition方法，决定哪些key落入哪些分区
  • 自定义group分组类，决定reduce阶段，哪些kv对，落入同一组，调用一次reduce()
  • 写main方法，设置自定义的类
    • job.setMapperClass
    • job.setPartitionerClass
    • job.setReducerClass
    • job.setGroupingComparatorClass

14. MapReduce数据倾斜（经常被问到）！！！

• 什么是数据倾斜？

  • 数据中不可避免地会出现离群值（outlier），并导致数据倾斜。这些离群值会显著地拖慢MapReduce的执行。

• 常见的数据倾斜有以下几类：

  • 数据频率倾斜——某一个区域的数据量要远远大于其他区域。比如某一个key对应的键值对远远大于其他键的键值对。
  • 数据大小倾斜——部分记录的大小远远大于平均值。

• 在map端和reduce端都有可能发生数据倾斜

  • 在map端的数据倾斜可以考虑使用combine
  • 在reduce端的数据倾斜常常来源于MapReduce的默认分区器

• 数据倾斜会导致map和reduce的任务执行时间大为延长，也会让需要缓存数据集的操作消耗更多的内存资源

14.1 如何诊断是否存在数据倾斜

2. 如何诊断哪些键存在数据倾斜？
   • 发现倾斜数据之后，有必要诊断造成数据倾斜的那些键。有一个简便方法就是在代码里实现追踪每个键的最大值。
   • 为了减少追踪量，可以设置数据量阀值，只追踪那些数据量大于阀值的键，并输出到日志中。实现代码如下
   • 运行作业后就可以从日志中判断发生倾斜的键以及倾斜程度；跟踪倾斜数据是了解数据的重要一步，也是设计MapReduce作业的重要基础

   package com.kaikeba.hadoop.dataskew;
   
   import org.apache.hadoop.io.IntWritable;
   import org.apache.hadoop.io.Text;
   import org.apache.hadoop.mapreduce.Reducer;
   import org.apache.log4j.Logger;
   
   import java.io.IOException;
   
   public class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
   
       private int maxValueThreshold;
   
       //日志类
       private static final Logger LOGGER = Logger.getLogger(WordCountReduce.class);
   
       @Override
       protected void setup(Context context) throws IOException, InterruptedException {
   
           //一个键达到多少后，会做数据倾斜记录
           maxValueThreshold = 10000;
       }
   
       /*
               (hello, 1)
               (hello, 1)
               (hello, 1)
               ...
               (spark, 1)
   
               key: hello
               value: List(1, 1, 1)
           */
       public void reduce(Text key, Iterable<IntWritable> values,
                             Context context) throws IOException, InterruptedException {
           int sum = 0;
           //用于记录键出现的次数
           int i = 0;
   
           for (IntWritable count : values) {
               sum += count.get();
               i++;
           }
   
           //如果当前键超过10000个，则打印日志
           if(i > maxValueThreshold) {
               LOGGER.info("Received " + i + " values for key " + key);
           }
   
           context.write(key, new IntWritable(sum));// 输出最终结果
       };
   }

14.2 减缓数据倾斜

• Reduce数据倾斜一般是指map的输出数据中存在数据频率倾斜的状况，即部分输出键的数据量远远大于其它的输出键

• 如何减小reduce端数据倾斜的性能损失？常用方式有：

  • 一、自定义分区

    • 基于输出键的背景知识进行自定义分区。

    • 例如，如果map输出键的单词来源于一本书。其中大部分必然是省略词（stopword）。那么就可以将自定义分区将这部分省略词发送给固定的一部分reduce实例。而将其他的都发送给剩余的reduce实例。

  • 二、Combine

    • 使用Combine可以大量地减小数据频率倾斜和数据大小倾斜。
    • combine的目的就是聚合并精简数据。

  • 三、抽样和范围分区

    • Hadoop默认的分区器是HashPartitioner，基于map输出键的哈希值分区。这仅在数据分布比较均匀时比较好。在有数据倾斜时就很有问题。

    • 使用分区器需要首先了解数据的特性。TotalOrderPartitioner中，可以通过对原始数据进行抽样得到的结果集来预设分区边界值。

    • TotalOrderPartitioner中的范围分区器可以通过预设的分区边界值进行分区。因此它也可以很好地用在矫正数据中的部分键的数据倾斜问题。

  • 四、数据大小倾斜的自定义策略

    • 在map端或reduce端的数据大小倾斜都会对缓存造成较大的影响，乃至导致OutOfMemoryError异常。处理这种情况并不容易。可以参考以下方法。

    • 设置mapreduce.input.linerecordreader.line.maxlength来限制RecordReader读取的最大长度。

    • RecordReader在TextInputFormat和KeyValueTextInputFormat类中使用。默认长度没有上限。

15. MR调优

• 有调优专题

16. 抽样、范围分区

16.1 数据

• 数据：气象站气象数据，来源美国国家气候数据中心（NCDC）（1901-2001年数据，每年一个文件）

  • 气候数据record的格式如下

16.2 需求

• 对气象数据，按照气温进行排序（气温符合正太分布）

16.3 实现方案

• 三种实现思路

  • 方案一：
    • 设置一个分区，即一个reduce任务；在一个reduce中对结果进行排序；
    • 失去了MR框架并行计算的优势
  • 方案二：
    • 自定义分区，人为指定各温度区间的记录，落入哪个分区；如分区温度边界值分别是-15、0、20，共4个分区
    • 但由于对整个数据集的气温分布不了解，可能某些分区的数据量大，其它的分区小，数据倾斜
  • 方案三：
    • 通过对键空间采样
    • 只查看一小部分键，获得键的近似分布（好温度的近似分布）
    • 进而据此结果创建分区，实现尽可能的均匀的划分数据集；
    • Hadoop内置了采样器；InputSampler

16.4 MR代码

分两大步

• 一、先将数据按气温对天气数据集排序。结果存储为sequencefile文件，气温作为输出键，数据行作为输出值

• 代码

  此代码处理原始日志文件

  结果用SequenceFile格式存储；

  温度作为SequenceFile的key；记录作为value

package com.kaikeba.hadoop.totalorder;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
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;

/**
 * 此代码处理原始日志文件 1901
 * 结果用SequenceFile格式存储；
 * 温度作为SequenceFile的key；记录作为value
 */
public class SortDataPreprocessor {

  //输出的key\value分别是气温、记录
  static class CleanerMapper extends Mapper<LongWritable, Text, IntWritable, Text> {
  
    private NcdcRecordParser parser = new NcdcRecordParser();
    private IntWritable temperature = new IntWritable();

    @Override
    protected void map(LongWritable key, Text value, Context context)
        throws IOException, InterruptedException {
      //0029029070999991901010106004+64333+023450FM-12+000599999V0202701N015919999999N0000001N9-00781+99999102001ADDGF108991999999999999999999
      parser.parse(value);
      if (parser.isValidTemperature()) {//是否是有效的记录
        temperature.set(parser.getAirTemperature());
        context.write(temperature, value);
      }
    }
  }


  //两个参数：/ncdc/input /ncdc/sfoutput
  public static void main(String[] args) throws Exception {

    if (args.length != 2) {
      System.out.println("<input> <output>");
    }

    Configuration conf = new Configuration();

    Job job = Job.getInstance(conf, SortDataPreprocessor.class.getSimpleName());
    job.setJarByClass(SortDataPreprocessor.class);
    //
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));

    job.setMapperClass(CleanerMapper.class);
    //最终输出的键、值类型
    job.setOutputKeyClass(IntWritable.class);
    job.setOutputValueClass(Text.class);
    //reduce个数为0
    job.setNumReduceTasks(0);
    //以sequencefile的格式输出
    job.setOutputFormatClass(SequenceFileOutputFormat.class);

    //开启job输出压缩功能
    //方案一
    conf.set("mapreduce.output.fileoutputformat.compress", "true");
    conf.set("mapreduce.output.fileoutputformat.compress.type","RECORD");
    //指定job输出使用的压缩算法
    conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.BZip2Codec");

    //方案二
    //设置sequencefile的压缩、压缩算法、sequencefile文件压缩格式block
    //SequenceFileOutputFormat.setCompressOutput(job, true);
    //SequenceFileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
    //SequenceFileOutputFormat.setOutputCompressorClass(job, SnappyCodec.class);
    //SequenceFileOutputFormat.setOutputCompressionType(job, SequenceFile.CompressionType.BLOCK);

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

• 二、全局排序

  使用全排序分区器TotalOrderPartitioner

package com.kaikeba.hadoop.totalorder;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
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.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;

import java.net.URI;

/**
 * 使用TotalOrderPartitioner全局排序一个SequenceFile文件的内容；
 * 此文件是SortDataPreprocessor的输出文件；
 * key是IntWritble，气象记录中的温度
 */
public class SortByTemperatureUsingTotalOrderPartitioner{

  /**
   * 两个参数：/ncdc/sfoutput /ncdc/totalorder
   * 第一个参数是SortDataPreprocessor的输出文件
   */
  public static void main(String[] args) throws Exception {
    if (args.length != 2) {
      System.out.println("<input> <output>");
    }

    Configuration conf = new Configuration();

    Job job = Job.getInstance(conf, SortByTemperatureUsingTotalOrderPartitioner.class.getSimpleName());
    job.setJarByClass(SortByTemperatureUsingTotalOrderPartitioner.class);

    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));

    //输入文件是SequenceFile
    job.setInputFormatClass(SequenceFileInputFormat.class);

    //Hadoop提供的方法来实现全局排序，要求Mapper的输入、输出的key必须保持类型一致
    job.setOutputKeyClass(IntWritable.class);
    job.setOutputFormatClass(SequenceFileOutputFormat.class);

    //分区器：全局排序分区器
    job.setPartitionerClass(TotalOrderPartitioner.class);

    //分了3个区；且分区i-1中的key小于i分区中所有的键
    job.setNumReduceTasks(3);

    /**
     * 随机采样器从所有的分片中采样
     * 每一个参数：采样率；
     * 第二个参数：总的采样数
     * 第三个参数：采样的最大分区数；
     * 只要numSamples和maxSplitSampled（第二、第三参数）任一条件满足，则停止采样
     */
    InputSampler.Sampler<IntWritable, Text> sampler =
            new InputSampler.RandomSampler<IntWritable, Text>(0.1, 5000, 10);
//    TotalOrderPartitioner.setPartitionFile();
    /**
     * 存储定义分区的键；即整个数据集中温度的大致分布情况；
     * 由TotalOrderPartitioner读取，作为全排序的分区依据，让每个分区中的数据量近似
     */
    InputSampler.writePartitionFile(job, sampler);

    //根据上边的SequenceFile文件（包含键的近似分布情况），创建分区
    String partitionFile = TotalOrderPartitioner.getPartitionFile(job.getConfiguration());
    URI partitionUri = new URI(partitionFile);

//    JobConf jobConf = new JobConf();

    //与所有map任务共享此文件，添加到分布式缓存中
    DistributedCache.addCacheFile(partitionUri, job.getConfiguration());
//    job.addCacheFile(partitionUri);

    //方案一：输出的文件RECORD级别，使用BZip2Codec进行压缩
    conf.set("mapreduce.output.fileoutputformat.compress", "true");
    conf.set("mapreduce.output.fileoutputformat.compress.type","RECORD");
    //指定job输出使用的压缩算法
    conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.BZip2Codec");

    //方案二
    //SequenceFileOutputFormat.setCompressOutput(job, true);
    //SequenceFileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
    //SequenceFileOutputFormat.setOutputCompressionType(job, CompressionType.BLOCK);

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

17 总结拓展

• 对大量数据进行全局排序

  • 先使用InputSampler.Sampler采样器，对整个key空间进行采样，得到key的近似分布

  • 保存到key分布情况文件中

  • 使用TotalOrderPartitioner，利用上边的key分布情况文件，进行分区；每个分区的数据量近似，从而防止数据倾斜

• 扩展阅读：《Hadoop权威指南（第4版）》

  • 7.3小节 - shuffle和排序

  • 8.2 输入格式——MR中还有一些自带的输入格式

  • 9.2.3 全排序

    
    

1. 描述MR的shuffle全流程（面试）
2. 谈谈什么是数据倾斜，什么情况会造成数据倾斜？（面试）
3. 对MR数据倾斜，如何解决？（面试）
4. 补充图解》》》》》》》》》》