使用Java API方式的MapReduce练习

ゝ一世哀愁。 2021-06-10 20:38 352阅读 0赞

众所周知，hadoop生态圈的多数组件都是使用java开发的。

那么使用Java API方式实现起来，显得要比其它语言效率更高，更原生态。

前面有一个[Hadoop学习笔记02\_MapReduce练习][Hadoop_02_MapReduce] 是在Linux下直接使用的python2.7实现的。这里我试试windows下用 java 来练习实现。

→\_→ 确认过眼神~~ 我是新手，感觉IDEA创建maven要比eclipse方便，更加好用。更主要的是，我在eclipse里找了半天没找到maven  >\_<|||

练习一：单词统计，wordcount

1. IntelliJ IDEA中New project, maven，SDK1.8 , Next,  输入 Groupid : examplemr , ArtifactId : examplemr , Version: 1.0 , Next,

Project name : examplemr , Project location: D:\\test\\examplemr   Finish

修改pom.xml  引入必要的dependency。 在IDE的提示中， 点击 Import Changes 等待自动下载相关的依赖包

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>cn.itcast</groupId>
        <artifactId>example-mr</artifactId>
        <version>1.0</version>
        <dependencies>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-common</artifactId>
                <version>2.7.4</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-hdfs</artifactId>
                <version>2.7.4</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-client</artifactId>
                <version>2.7.4</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-mapreduce-client-core</artifactId>
                <version>2.7.4</version>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    
                    <artifactId>maven-jar-plugin</artifactId>
                    <version>2.4</version>
                    <configuration>
                        <archive>
                            <manifest>
                                <addClasspath>true</addClasspath>
                                <classpathPrefix>lib/</classpathPrefix>
                                <mainClass>cn.itcast.hadoop.mr.WordCountDriver</mainClass>
                            </manifest>
                        </archive>
                    </configuration>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.0</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                        <encoding>UTF-8</encoding>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
    </project>

展开src,main,java, 右击, new,package,输入cn.abc.hadoop.mr  再 New, Java Class, 输入WordCountMapper

具体代码如下， 注意import时的包名正确。

package cn.abc.hadoop.mr;
    
    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;
    
    /**
     * Created by abc .
     *
     * 这里就是mapreduce程序  mapper阶段业务逻辑实现的类
     *
     * Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
     *
     * KEYIN：表示mapper数据输入的时候key的数据类型，在默认的读取数据组件下，叫InputFormat,它的行为是一行一行的读取待处理的数据
     *        读取一行，返回一行给我们的mr程序，这种情况下  keyin就表示每一行的起始偏移量  因此数据类型是Long
     *
     * VALUEIN:表述mapper数据输入的时候value的数据类型，在默认的读取数据组件下 valuein就表示读取的这一行内容  因此数据类型是String
     *
     * KEYOUT 表示mapper数据输出的时候key的数据类型  在本案例当中 输出的key是单词  因此数据类型是 String
     *
     * VALUEOUT表示mapper数据输出的时候value的数据类型  在本案例当中 输出的key是单词的次数  因此数据类型是 Integer
     *
     * 这里所说的数据类型String Long都是jdk自带的类型   在序列化的时候  效率低下 因此hadoop自己封装一套数据类型
     *   long---->LongWritable
     *   String-->Text
     *   Integer--->Intwritable
     *   null-->NullWritable
     *
     *
     */
    public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
    
        /**
         *  这里就是mapper阶段具体的业务逻辑实现方法  该方法的调用取决于读取数据的组件有没有给mr传入数据
         *  如果有的话  每传入一个《k,v》对  该方法就会被调用一次
         * @param key
         * @param value
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    
            //拿到传入进来的一行内容，把数据类型转化为String
            String line = value.toString();
    
            //将这一行内容按照分隔符进行一行内容的切割 切割成一个单词数组
            String[] words = line.split(" ");
    
            //遍历数组，每出现一个单词  就标记一个数字1  <单词，1>
            for (String word : words) {
                //使用mr程序的上下文context 把mapper阶段处理的数据发送出去
                //作为reduce节点的输入数据
                context.write(new Text(word),new IntWritable(1));
                //hadoop hadoop spark -->   <hadoop,1><hadoop,1><spark,1>
            }
        }
    }

继续 New, Java Class, WordCountReducer 代码如下：

package cn.abc.hadoop.mr;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    /**
     * Created by abc
     *
     * 这里是MR程序 reducer阶段处理的类
     *
     * KEYIN：就是reducer阶段输入的数据key类型，对应mapper的输出key类型  在本案例中  就是单词  Text
     *
     * VALUEIN就是reducer阶段输入的数据value类型，对应mapper的输出value类型  在本案例中  就是单词次数  IntWritable
     * .
     * KEYOUT就是reducer阶段输出的数据key类型 在本案例中  就是单词  Text
     *
     * VALUEOUTreducer阶段输出的数据value类型 在本案例中  就是单词的总次数  IntWritable
     */
    public class WordCountReducer extends Reducer<Text,IntWritable,Text,IntWritable> {
    
        /**
         * 这里是reduce阶段具体业务类的实现方法
         * @param key
         * @param values
         * @param context
         * @throws IOException
         * @throws InterruptedException
         *
         * reduce接收所有来自map阶段处理的数据之后，按照key的字典序进行排序
         * <hello,1><hadoop,1><spark,1><hadoop,1>
         * 排序后：
         * <hadoop,1><hadoop,1><hello,1><spark,1>
         *
         *按照key是否相同作为一组去调用reduce方法
         * 本方法的key就是这一组相同kv对的共同key
         * 把这一组所有的v作为一个迭代器传入我们的reduce方法
         *
         * <hadoop,[1,1]>
         */
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            //定义一个计数器
            int count = 0;
            //遍历一组迭代器，把每一个数量1累加起来就构成了单词的总次数
    
            for(IntWritable value:values){
                count +=value.get();
            }
    
            //把最终的结果输出
            context.write(key,new IntWritable(count));
        }
    }

最后，WordCountDriver 代码，写完就可以右击，Run 一下试试。

package cn.abc.hadoop.mr;
    
    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;
    
    /**
     * Created by abc .
     *
     * 这个类就是mr程序运行时候的主类，本类中组装了一些程序运行时候所需要的信息
     * 比如：使用的是那个Mapper类  那个Reducer类  输入数据在那 输出数据在什么地方
     */
    public class WordCountDriver {
        public static void main(String[] args) throws Exception{
            //通过Job来封装本次mr的相关信息
            Configuration conf = new Configuration();
            // 即使没有下面这行,也可以本地运行 因\hadoop-mapreduce-client-core-2.7.4.jar!\mapred-default.xml 中默认的参数就是 local
            //conf.set("mapreduce.framework.name","local");
            Job job = Job.getInstance(conf);
    
            //指定本次mr job jar包运行主类
            job.setJarByClass(WordCountDriver.class);
    
            //指定本次mr 所用的mapper reducer类分别是什么
            job.setMapperClass(WordCountMapper.class);
            job.setReducerClass(WordCountReducer.class);
    
            //指定本次mr mapper阶段的输出  k  v类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
    
            //指定本次mr 最终输出的 k v类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            // job.setNumReduceTasks(3); //ReduceTask个数
    
            //如果业务有需求，就可以设置combiner组件
            job.setCombinerClass(WordCountReducer.class);
    
            //指定本次mr 输入的数据路径 和最终输出结果存放在什么位置
            FileInputFormat.setInputPaths(job,"D:\\wordcount\\input");
            FileOutputFormat.setOutputPath(job,new Path("D:\\wordcount\\output"));
            //如果出现0644错误或找不到winutils.exe,则需要设置windows环境和相关文件.
    
            //上面的路径是本地测试时使用，如果要打包jar到hdfs上运行时，需要使用下面的路径。
            //FileInputFormat.setInputPaths(job,"/wordcount/input");
            //FileOutputFormat.setOutputPath(job,new Path("/wordcount/output"));
    
            // job.submit(); //一般不要这个.
            //提交程序  并且监控打印程序执行情况
            boolean b = job.waitForCompletion(true);
            System.exit(b?0:1);
        }
    }

如果出现0644错误或找不到winutils.exe,则需要设置windows环境和相关文件.

设置环境变量HADOOP\_HOME指向hadoop-2.7.5.tar.gz完整包解压后的路径，并且在Path中添加 %HADOOP\_HOME%\\bin;

然后下载win环境所需文件，解压，将其复制到环境变量为HADOOP\_HOME的真实路径下的bin目录中。

如果想要看看日志，可以在 examplemr\\src\\main\\resources\\下放入 log4j.properties 文件。运行后，将会看到\\examplemr\\mapreduce\_test.log 日志。

相关文件： [hadoop-win相关文件winutils.exe][hadoop-win_winutils.exe] | [log4j.properties][]

如果要打成jar包，传到HDFS上运行：

需要在pom.xml中添加 build 部分，详见上面pom.xml中的 build部分，注意主类入口就是examplemr\\src\\main\\java\\cn\\abc\\hadoop\\mr\\下的 WordCountDriver.java

<mainClass>cn.abc.hadoop.mr.WordCountDriver</mainClass>

且需要修改 WordCountDriver.java中的路径信息：

//如果要打包jar到hdfs上运行时，需要使用下面的路径。

FileInputFormat.setInputPaths(job,"/wordcount/input");
    FileOutputFormat.setOutputPath(job,new Path("/wordcount/output"));

在IDEA中, View, Tool Windows, Maven Project 显示出工具，双击项目下Lifecycle中的 package 等待打包完成。

然后找到项目中 target下，example-mr-1.0.jar文件，传入Linux，

然后在hadoop已成功启动的前提下，执行：

hdfs dfs -mkdir -p /wordcount/input  # 创建对应的文件夹
    
    hdfs dfs -put 1.txt 2.txt /wordcount/input/  # 上传需要分析的数据文件
    
    hadoop jar example-mr-1.0.jar  # 运行此MR程序。因为打包前在pom.xml中指定了主类入口，所以，这里不再指定入口参数。

等待执行结果，或者上 http://hadoop主机名:8088上查看进程与结果。

练习二：简单流量统计，FlowSum

首先是如下这样的日志文件：需要统计倒数第2(上行流量)、倒数第3(下行流量) 相加后的总流量，条件是基于手机号码(第2字段)

1363157985066	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com	24	27	2481	24681	200
    1361157912132	13726230513	00-FD-07-A4-72-B8:CMCC	120.196.40.8	4	7	248	0	200
    1363157985033	13826230523	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com	24	27	2481	24681	200
    1363157985012	13726230533	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com	24	27	2481	24681	200
    1363157125076	13726230543	00-FD-07-A4-72-B8:CMCC	120.196.100.82	视频网站	24	27	1527	2106	200
    1363157985011	13926230553	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com	24	27	2481	24681	200
    1363157985016	13826230563	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com	24	27	2481	24681	200
    1363157985123	13926230573	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com	24	27	2481	24681	200
    1363157985135	18912688533	00-FD-07-A4-72-B8:CMCC	220.196.100.82	综合门户	15	12	1938	2910	200
    1363157985432	18912688533	00-FD-07-A4-72-B8:CMCC	220.196.100.82	i02.c.aliimg.com	24	27	3333	21321	200
    1363157985321	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	搜索引擎	24	27	9531	9531	200
    1363157985222	13826230523	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com	24	27	2481	24681	200
    1363157983331	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com	24	27	2481	24681	200

pom.xml文件：

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>cn.itcast</groupId>
        <artifactId>example-mr</artifactId>
        <version>1.0</version>
        <dependencies>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-common</artifactId>
                <version>2.7.4</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-hdfs</artifactId>
                <version>2.7.4</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-client</artifactId>
                <version>2.7.4</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-mapreduce-client-core</artifactId>
                <version>2.7.4</version>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    
                    <artifactId>maven-jar-plugin</artifactId>
                    <version>2.4</version>
                    <configuration>
                        <archive>
                            <manifest>
                                <addClasspath>true</addClasspath>
                                <classpathPrefix>lib/</classpathPrefix>
                                <mainClass>cn.abc.hadoop.mr.WordCountDriver</mainClass>
                            </manifest>
                        </archive>
                    </configuration>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.0</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                        <encoding>UTF-8</encoding>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
    </project>

Map阶段:

package cn.abc.hadoop.flowsum;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    public class FlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean>{
    
        Text k = new Text();
        FlowBean v = new FlowBean(); // new一次对象,多次使用. 效率较高
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split("\t"); //数据文件如果是两个连续的 \t 会有问题.
    
            String phoneNum = fields[1];
            long upFlow = Long.parseLong(fields[fields.length-3]); //倒着获取需要的字段
            long downFlow = Long.parseLong(fields[fields.length-2]);
    
            // context.write(new Text(phoneNum),new FlowBean(upFlow,downFlow)); //效率太低,每次都产生对象.
    
            //使用对象中的set方法来写入数据,避免大量new对象
            k.set(phoneNum);
            v.set(upFlow,downFlow);
            context.write(k,v);
        }
    }

Reduce阶段：

package cn.abc.hadoop.flowsum;
    
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    /**
     * <手机号1, bean><手机号2, bean><手机号2, bean><手机号, bean>
     *
     *     <手机号1, bean><手机号1, bean>
     *     <手机号2, bean><手机号2, bean>
     */
    public class FlowSumReducer extends Reducer<Text,FlowBean,Text,FlowBean>{
    
        FlowBean v = new FlowBean();
        @Override
        protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
            long upFlowCount = 0;
            long downFlowCount = 0;
    
            for(FlowBean bean:values){
                upFlowCount += bean.getUpFlow();
                downFlowCount += bean.getDownFlow();
            }
            v.set(upFlowCount,downFlowCount);
            context.write(key,v);
        }
    }

驱动： 还是先在本地测试，写本地路径。

package cn.abc.hadoop.flowsum;
    
    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;
    
    public class FlowSumDriver {
        public static void main(String[] args) throws Exception{
            //通过Job来封装本次mr的相关信息
            Configuration conf = new Configuration();
            //conf.set("mapreduce.framework.name","local");
            Job job = Job.getInstance(conf);
    
            //指定本次mr job jar包运行主类
            job.setJarByClass(FlowSumDriver.class);
    
            //指定本次mr 所用的mapper reducer类分别是什么
            job.setMapperClass(FlowSumMapper.class);
            job.setReducerClass(FlowSumReducer.class);
    
            //指定本次mr mapper阶段的输出  k  v类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(FlowBean.class);
    
            //指定本次mr 最终输出的 k v类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
    
            //指定本次mr 输入的数据路径 和最终输出结果存放在什么位置
            FileInputFormat.setInputPaths(job,"D:\\flowsum\\input");
            FileOutputFormat.setOutputPath(job,new Path("D:\\flowsum\\output"));
            //FileInputFormat.setInputPaths(job,"/wordcount/input");
            //FileOutputFormat.setOutputPath(job,new Path("/wordcount/output"));
    
            // job.submit(); //一般不要这个.
            //提交程序  并且监控打印程序执行情况
            boolean b = job.waitForCompletion(true);
            System.exit(b?0:1);
        }
    }

用到的类：

package cn.abc.hadoop.flowsum;
    
    import org.apache.hadoop.io.Writable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    public class FlowBean implements Writable {
        //写上这句{}之后,点击Writable,点击左边的灯泡，Implement Mothods 选中write和readFields后OK。
        // Hadoop的Writable 比JAVA自带的要高效很多。
        //手工输入，定义三个属性
        private long upFlow;
        private long downFlow;
        private long sumFlow;
    
        // R click blank, Generator , Constructor, Select None 生成无参构造函数
        public FlowBean() {
        }
    
        //右击类名,Generate, Constructor,选中所有子项，OK，生成有参构造函数
        // R click blank, Generator , Constructor, select All, OK
        public FlowBean(long upFlow, long downFlow, long sumFlow) {
            this.upFlow = upFlow;
            this.downFlow = downFlow;
            this.sumFlow = sumFlow;
        }
    
        //自己构造一个2个参数的方法。
        public FlowBean(long upFlow, long downFlow) {
            this.upFlow = upFlow;
            this.downFlow = downFlow;
            this.sumFlow = upFlow + downFlow;
        }
    
        //自己构造一个有返回值的 set方法，
        public void set(long upFlow, long downFlow) {
            this.upFlow = upFlow;
            this.downFlow = downFlow;
            this.sumFlow = upFlow + downFlow;
        }
    
        //这是序列化方法
        @Override
        public void write(DataOutput out) throws IOException {
            out.writeLong(upFlow);
            out.writeLong(downFlow);
            out.writeLong(sumFlow);
        }
    
        @Override
        public String toString() {
            return upFlow+"\t"+downFlow+"\t"+sumFlow;
        }
    
        //这是反序列化方法
        //反序列化时候, 注意序列化的顺序
        // 先序列化的先出来.
        @Override
        public void readFields(DataInput in) throws IOException {
            this.upFlow = in.readLong();
            this.downFlow = in.readLong();
            this.sumFlow = in.readLong();
        }
    
        // 右击类名,Generate, Getter and Setter, 选中全部，OK
        public long getUpFlow() {
            return upFlow;
        }
    
        public void setUpFlow(long upFlow) {
            this.upFlow = upFlow;
        }
    
        public long getDownFlow() {
            return downFlow;
        }
    
        public void setDownFlow(long downFlow) {
            this.downFlow = downFlow;
        }
    
        public long getSumFlow() {
            return sumFlow;
        }
    
        public void setSumFlow(long sumFlow) {
            this.sumFlow = sumFlow;
        }
    }

最后在output文件夹下 part-r-00000 中得到类似于下面的结果：

![20200304155603637.png][]

又碰到改需求的情况了， 现在要把前面得到的统计结果按总流量降序排列

也就是在前一次MR的基础上，再来一次MR。

在前面flowsum包下再建一个包 sort。

修改FlowBean主类为FlowBean implements WritableComparable<FlowBean> 并且实现public int compareTo 方法

import org.apache.hadoop.io.WritableComparable;
    
    
    public class FlowBean implements WritableComparable<FlowBean> {
        //写上WritableComparable，右击，Generate, Implement Mothods, compareTo, OK。
    
        // 原有代码不变... ...
    
        //这里就是Bean比较大小的方法  默认是如果指定的数与参数相等返回0，指定的数小于参数返回-1, 大于返回1
        @Override
        public int compareTo(FlowBean o) { //来自主类中方法的重写
            //实现按照总流量的倒序排序
            return this.sumFlow > o.getSumFlow() ? -1 : 1;
            //正常逻辑 return this.sumFlow > o.getSumFlow() ? 1 : -1;
        }

在sort包下添加类，FlowSumSort

package cn.abc.hadoop.flowsum.sort;
    
    import cn.abc.hadoop.flowsum.FlowBean;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    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;
    
    /**
     * 又碰到改需求的情况了， 现在要把前面 FlowSum 得到的统计结果按总流量降序排列
     */
    public class FlowSumSort {
        public static class FlowSumSortMapper extends Mapper<LongWritable,Text,FlowBean,Text>{
    
            Text v = new Text();
            FlowBean k = new FlowBean();
    
            protected void map(LongWritable key,Text value,Context context) throws IOException,InterruptedException{
                String line = value.toString();
                String[] fields = line.split("\t");
                String phoneNum = fields[0];
    
                long upFlow = Long.parseLong(fields[1]);
                long downFlow = Long.parseLong(fields[2]);
    
                k.set(upFlow,downFlow);
                v.set(phoneNum);
    
                context.write(k,v);
            }
        }
    
    
        public static class FlowSumSortReducer extends Reducer<FlowBean,Text,Text,FlowBean>{
            @Override
            protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
                context.write(values.iterator().next(),key);
            }
        }
    
    
        public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
    
            //指定这个job所在的jar包位置
            job.setJarByClass(FlowSumSort.class);
    
            //指定使用的Mapper是哪个类，Reducer是哪个类
            job.setMapperClass(FlowSumSortMapper.class);
            job.setReducerClass(FlowSumSortReducer.class);
    
            //设置业务逻辑Mapper类的输出key 和value的数据类型
            job.setMapOutputKeyClass(FlowBean.class);
            job.setMapOutputValueClass(Text.class);
    
            //设置业务逻辑Reducer类的输出key 和value的数据类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
    
            FileInputFormat.setInputPaths(job,"D:\\flowsum\\output"); //使用上一次MR的输出结果为输入数据
            //指定处理完成之后的结果保存位置
            FileOutputFormat.setOutputPath(job,new Path("D:\\flowsum\\outputsort"));
    
            //向yarn集群提交这个job
            boolean res = job.waitForCompletion(true);
            System.exit(res ? 0 : 1);
        }
    }

最后执行，看看得到的结果outputsort是不是已经按照总流量降序排列。

13726230503	14493	58893	73386
    13826230523	4962	49362	54324
    18912688533	5271	24231	29502
    13726230533	2481	24681	27162
    13826230563	2481	24681	27162
    13926230553	2481	24681	27162
    13926230573	2481	24681	27162
    13726230543	1527	2106	3633
    13726230513	248	0	248

继续改需求， 现在要按手机归属地进行统计

修改重新来一次MR。

在flowsum下新建包，partitioner, 添加类ProvincePartitioner.java

package cn.abc.hadoop.flowsum.partitioner;
    
    import cn.abc.hadoop.flowsum.FlowBean;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    import java.util.HashMap;
    
    
    public class ProvincePartitioner extends Partitioner<Text, FlowBean> {
        public static HashMap<String, Integer> provinceMap = new HashMap();
    
        static {
            provinceMap.put("137", 0); //模拟手机归属地
            provinceMap.put("138", 1);
            provinceMap.put("139", 2);
        }
    
        //这里就是实际分区方法，返回分区编号，分区编号就决定了数据到哪个分区中
        //第三个参数 numPartitions 应该就是设置的 numReuduceTasks个数
        //默认分区参数可参考：HashPartitioner
        @Override
        public int getPartition(Text key, FlowBean value, int numPartitions) {
            Integer code = provinceMap.get(key.toString().substring(0, 3));
    
            if (code != null) {
                return code;
            }
    
            return 3; //不在上方列表中的
        }
    }

MR程序写在一个文件中 : FlowSumProvince

package cn.abc.hadoop.flowsum.partitioner;
    
    import cn.abc.hadoop.flowsum.FlowBean;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    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;
    
    /**
     * 改需求：将流量统计结果按手机归属地不同省份输出到不同文件中。
     */
    public class FlowSumProvince {
        public static class FlowSumProvinceMapper extends Mapper<LongWritable, Text,Text, FlowBean> {
    
            Text k = new Text();
            FlowBean v = new FlowBean();
    
            protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
                //拿取一行文本转为String
                String line = value.toString();
                //按照分隔符 \t 进行分隔
                String[] fields = line.split("\t");
                //获取用户手机号
                String phoneNum = fields[1];
    
                long upFlow = Long.parseLong(fields[fields.length - 3]);
                long downFlow = Long.parseLong(fields[fields.length - 2]);
    
                k.set(phoneNum);
                v.set(upFlow, downFlow);
    
                context.write(k, v);
            }
        }
    
    
        public static class FlowSumProvinceReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
    
            FlowBean v = new FlowBean();
    
            @Override
            protected void reduce(Text key, Iterable<FlowBean> flowBeans, Context context) throws IOException, InterruptedException {
    
                long upFlowCount = 0;
                long downFlowCount = 0;
    
                for (FlowBean flowBean : flowBeans){
                    upFlowCount += flowBean.getUpFlow();
                    downFlowCount += flowBean.getDownFlow();
                }
    
                v.set(upFlowCount,downFlowCount);
                context.write(key,v);
            }
        }
    
    
        public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
    
            //指定这个job所在的jar包位置
            job.setJarByClass(FlowSumProvince.class);
    
            //指定使用的Mapper是哪个类，Reducer是哪个类
            job.setMapperClass(FlowSumProvinceMapper.class);
            job.setReducerClass(FlowSumProvinceReducer.class);
    
            //设置业务逻辑Mapper类的输出key 和value的数据类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(FlowBean.class);
    
            //设置业务逻辑Reducer类的输出key 和value的数据类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
    
            job.setNumReduceTasks(4); //设定ReduceTask数量 等于ProvincePartitioner中的provinceMap中数量
            // 分区个数 < ReduceTasks个数，正常执行，会有空结果文件产生
            // 分区个数 > ReduceTasks个数，错误 Illegal partition
    
    
            //这里指定自定义分区组件，如果不指定，默认就是hashcode
            job.setPartitionerClass(ProvincePartitioner.class);
    
            FileInputFormat.setInputPaths(job, "D:\\flowsum\\input"); //输入数据
            //指定处理完成之后的结果保存位置
            FileOutputFormat.setOutputPath(job, new Path("D:\\flowsum\\outputProvince"));
    
            //向yarn集群提交这个job
            boolean res = job.waitForCompletion(true);
            System.exit(res ? 0 : 1);
        }
    }

因为这次指定了分区组件ProvincePartitioner ，所以，得到了4个结果文件。也就是按照ProvincePartitioner里面的"137","138","139","其它"分区的结果。

**注：经过多次尝试后发现，在非hadoop环境下运行main函数，partitionClass中的返回分区值，取决于reduceTasks的取值范围，比如：numReduceTasks=3，返回值需要在区间\[0,2\]内，如果只有返回0，2的话，还是会生成part0、part1、part2三个文件，只不过part1会是一个空文件，里面没有内容，part0与part2中是正常的数据输出。**

查看默认的PartitionClass后发现，这个数字的取值范围其实就是 对 设置的 NumReduceTasks的个数 取模，如果task个数设置为5，则PartitionCLass的分区返回只能在0-5 之间

默认分区类：

package org.apache.hadoop.mapreduce.lib.partition;
    
    import org.apache.hadoop.classification.InterfaceAudience.Public;
    import org.apache.hadoop.classification.InterfaceStability.Stable;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    @Public
    @Stable
    public class HashPartitioner<K, V> extends Partitioner<K, V> {
        public HashPartitioner() {
        }
    
        public int getPartition(K key, V value, int numReduceTasks) {
            return (key.hashCode() & 2147483647) % numReduceTasks;
        }
    }

转自：[https://www.iteye.com/blog/chengjianxiaoxue-2164473][https_www.iteye.com_blog_chengjianxiaoxue-2164473]

a) 默认下分配一个区

b) 分配几个区，则对应几个reduce任务，每个任务在执行的时候都会公用reduce内的代码

c) 自定义分区下 返回的分区数量一定要和 定义的reduce任务相同，具体来说就是：

自定义分区类 extends HashPartitioner，重写getPartition时，返回的分支个数要和

job.setNumReduceTasks(X); 中的X个数相同

如果分区格式和reducetask任务个数不同下，在hadoop不同版本中的运行情况如下：

Java代码

1.  HashPartitioner.java  key.hashcode() & integer.maxvalue % numreducetasks =  模1恒等于0 返回值恒为0 返回值是分区的标记或者索引 part-00000 part-00001 等等  
2.  默认的是job.setPartitionerClass(HashPartitioner.class)  自定义分区返回的是索引数字，从0开始依次递增1返回。  
3.  以 手机号和座机号写在一个文件中为例：  
4.  如果分区数量 大于/小于 reduce数量时,   
    
    1.  2个分区 1个reduce --->  hadoop2中依旧能正常执行 只不过不会分区 所有数据都写到一个输出中   hadoop1中会报错 
    2.  2个分区 4个reduce --->  hadoop2中依旧能正常执行 输出结果写到4个区中，第一个分区结果为手机号 第二个为座机号 剩下两个为空文件 所有数据都写到一个输出中

![format_png][]

[Hadoop_02_MapReduce]: https://www.cnblogs.com/frx9527/p/mapreduce.html
[hadoop-win_winutils.exe]: https://files.cnblogs.com/files/frx9527/hadoop-win.rar
[log4j.properties]: https://files.cnblogs.com/files/frx9527/log4j.rar
[20200304155603637.png]: /images/20210527/bc297aa563e44ab48f50c466ea22c0df.png
[https_www.iteye.com_blog_chengjianxiaoxue-2164473]: https://www.iteye.com/blog/chengjianxiaoxue-2164473
[format_png]: /images/20210527/5c11152d2783448e9ac305224aa26695.png