logo

小白学苑

让大数据学习更简单

MapReduce基础案例09-自定义分区器

什么是分区器Partitioner？

分区器(Partitioner)，是用来将mapper的中间输出结果key/value对进行分区的类。分到同一个分区的key/value对会被shuffle到同一个reducer上。

partition阶段发生在Map阶段之后Reduce阶段之前。分区的数量等于reducer的数量，这意味着分区器将根据reducer的数量对数据进行分组。

一个Partitioner实例与Mapper实例执行在同一个JVM中。 在Mapper实例中每执行一次context.write，就调用一次Partitioner的getPartition方法。getPartition方法的签名如下：

int getPartition(K key, V value, int numReduceTasks)

默认情况下，Hadoop框架使用下面这个类对key进行哈希分区：

org.apache.hadoop.mapreduce.lib.partition.HashPartitioner

有的时候，用户希望使用用户自定义的条件进行分区，而不是使用默认的分区方式（默认按key进行哈希分区）， 这个时候就需要自已来定义分区器类。用户自定义分区类需要继承自org.apache.hadoop.mapreduce.Partitioner类。

问题描述

现在有如下的员工信息：


这些信息存储在数据文件input.txt中，如下：

1201,gopal,45,Male,50000
1202,manisha,40,Female,51000
1203,khaleel,34,Male,30000
1204,prasanth,30,Male,31000
1205,kiran,20,Male,40000
1206,laxmi,25,Female,35000
1207,bhavya,20,Female,15000
1208,reshma,19,Female,14000
1209,kranthi,22,Male,22000
1210,Satish,24,Male,25000
1211,Krishna,25,Male,26000
1212,Arshad,28,Male,20000
1213,lavanya,18,Female,8000

要求编写MapReduce应用程序，处理输入数据集，按性别找出不同年龄组中最高薪水的员工 (例如, 小于20岁, 21岁至30岁之间, 大于30岁)，并输出。


一、创建Java Maven项目

Maven依赖：

<?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>HadoopDemo</groupId>
    <artifactId>com.xueai8</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <!--hadoop依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.3.1</version>
        </dependency>
        <!--hdfs文件系统依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.3.1</version>
        </dependency>
        <!--MapReduce相关的依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
            <version>3.3.1</version>
        </dependency>
        <!--junit依赖-->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!--编译器插件用于编译拓扑-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <!--指定maven编译的jdk版本和字符集,如果不指定,maven3默认用jdk 1.5 maven2默认用jdk1.3-->
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source> <!-- 源代码使用的JDK版本 -->
                    <target>1.8</target> <!-- 需要生成的目标class文件的编译版本 -->
                    <encoding>UTF-8</encoding><!-- 字符集编码 -->
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

PartMapper.java：

package com.xueai8.partition;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;

public class PartMapper extends Mapper<LongWritable, Text, Text, Text> {

    // 定义可重用的key和value
    private static Text gender = new Text();

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] str = value.toString().split(",");     // 分词
        gender.set(str[3]);                 // 取性别作为key
        context.write(gender, value);       // 写出k-v对
    }
}

CaderPartitioner.java：自定义分区器类

package com.xueai8.partition;

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

/**
 *
 * 自定义分区器
 * 根据给定的分区条件，输入的key-value对数据基于年龄条件可以被分为三部分
 */
public class CaderPartitioner extends Partitioner<Text, Text> {

    // 输入数据是map的输出
    @Override
    public int getPartition(Text key, Text value, int numReduceTasks) {
        String[] str = value.toString().split(",");    // 对员工信息分词
        int age = Integer.parseInt(str[2]);            // 取年龄字段

        // 根据条件，返回不同的分区id
        if (numReduceTasks == 0) {
            return 0;   // 0
        }
        if (age <= 20) {
            return 0;   // 0
        } else if (age <= 30) {
            return 1;   // 1
        } else {
            return 2;   // 2
        }
    }
}

PartReducer.java

package com.xueai8.partition;

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

public class PartReducer extends Reducer<Text, Text, Text, IntWritable> {

    private static IntWritable value = new IntWritable(0);

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException,
            InterruptedException {
        int max = -1;
        for (Text val : values) {
            String[] str = val.toString().split(",");       // 对每一行分词
            int salary = Integer.parseInt(str[4]);          // 取薪资字段
            if ( salary > max) {
                max = salary;
            }
        }
        value.set(max);
        context.write(new Text(key), value);
    }
}

PartDriver.java：

package com.xueai8.partition;

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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import java.io.IOException;

public class PartDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        if (args.length < 2) {
            System.err.println("用法： <input_path> <output_path>");
            System.exit(-1);
        }

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "PartitionerDemo");
        job.setJarByClass(PartDriver.class);

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

        // set mapper
        job.setMapperClass(PartMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        /*** 注意：这里指定使用自定义的分区器 ***/
        job.setPartitionerClass(CaderPartitioner.class);

        // set reducer
        job.setReducerClass(PartReducer.class);
        job.setNumReduceTasks(3);   /*** 相应要指定3个reducer ****/

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        System.out.println(job.waitForCompletion(true) ? 0 : -1);
    }
}

二、配置log4j

在src/main/resources目录下新增log4j的配置文件log4j.properties，内容如下：

log4j.rootLogger = info,stdout

### 输出信息到控制抬 ###
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n

三、项目打包

打开IDEA下方的终端窗口terminal，执行"mvn clean package"打包命令，如下图所示：

如果一切正常，会提示打jar包成功。如下图所示：

这时查看项目结构，会看到多了一个target目录，打好的jar包就位于此目录下。如下图所示：

四、项目部署

请按以下步骤执行。

1、启动HDFS集群和YARN集群。在Linux终端窗口中，执行如下的脚本：

    $ start-dfs.sh
    $ start-yarn.sh

查看进程是否启动，集群运行是否正常。在Linux终端窗口中，执行如下的命令：

    $ jps

这时应该能看到有如下5个进程正在运行，说明集群运行正常：

    5542 NodeManager
    5191 SecondaryNameNode
    4857 NameNode
    5418 ResourceManager
    4975 DataNode

2、将数据文件sample.txt上传到HDFS的/data/mr/目录下。

$ hdfs dfs -mkdir -p /data/mr
$ hdfs dfs -put input.txt /data/mr/
$ hdfs dfs -ls /data/mr/

3、提交作业到Hadoop集群上运行。（如果jar包在Windows下，请先拷贝到Linux中。）

在终端窗口中，执行如下的作业提交命令：

$ hadoop jar com.xueai8-1.0-SNAPSHOT.jar com.xueai8.partition.PartDriver /data/mr /data/mr-output

4、查看输出结果。

在终端窗口中，执行如下的HDFS命令，查看输出结果：

$ hdfs dfs -ls /data/mr-output

会发现生成了三个输出结果文件，每个reducer(每个分区)对应一个输出文件。

查看第一个结果文件中的内容：

$ hdfs dfs -cat /data/mr-output/part-r-00000

可以看到如下的输出结果：

Female	15000
Male	40000

查看第二个结果文件中的内容：

$ hdfs dfs -cat /data/mr-output/part-r-00001

可以看到如下的输出结果：

Female	35000
Male	31000

查看第三个结果文件中的内容：

$ hdfs dfs -cat /data/mr-output/part-r-00002

可以看到如下的输出结果：

Female	51000
Male	50000
《PySpark原理深入与编程实战》