深入浅出学大数据（四）MapReduce快速入门及其编程实践

map: (k1; v1) → [(k2; v2)]
reduce: (k2; [v2]) → [(k3; v3)]

public FlowBean() {
	super();
}

@Override
public void write(DataOutput out) throws IOException {
	out.writeLong(upFlow);
	out.writeLong(downFlow);
	out.writeLong(sumFlow);
}

@Override
public void readFields(DataInput in) throws IOException {
	upFlow = in.readLong();
	downFlow = in.readLong();
	sumFlow = in.readLong();
}

@Override
public int compareTo(FlowBean o) {
	// 倒序排列，从大到小
	return this.sumFlow > o.getSumFlow() ? -1 : 1;
}

id	手机号码		网络ip						上行流量  下行流量     网络状态码
1	13736230513	192.196.100.1	www.Jayce.com	2481	24681	200
8 	15910133277	192.168.100.5	www.hao123.com	3156	2936	200
12 	15959002129	192.168.100.9	www.Jayce.com	1938	180		500
....

手机号码		    上行流量        下行流量		总流量
13590439668 		1116		      954 			2070

hadoop@Master:/opt/moudle/hadoop/myapp$ hadoop jar MapReduceDemo.jar WordCountDriver input	output

hadoop@Master:/opt/moudle/hadoop/myapp$ yarn jar MapReduceDemo.jar com.buwenbuhuo.mapreduce.WordCount.WordCountDriver input output

        // 设置自定义分区器
        job.setPartitionerClass(ProvincePartitioner1.class);
        // 设置对应的ReduceTask的个数
        job.setNumReduceTasks(5);

public class HashPartitioner<K, V> extends Partitioner<K, V> {

  /** Use {@link Object#hashCode()} to partition. */
  public int getPartition(K key, V value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }
}

hadoop@Master:~$ sudo apt-get install openssl
hadoop@Master:~$ sudo apt-get install libssl-dev

<property>
          <name>mapreduce.map.output.compress</name>
          <value>true</value>
</property>
<property>
         <name>mapreduce.map.output.compress.codec</name>
         <value>org.apache.hadoop.io.compress.SnappyCodec</value>
</property>

<property>       
				<name>mapreduce.output.fileoutputformat.compress</name>
       						<value>true</value>
</property>
<property>       <name>mapreduce.output.fileoutputformat.compress.type</name>
       							 <value>RECORD</value>
</property>
<property>        
 				<name>mapreduce.output.fileoutputformat.compress.codec</name>
        						<value>org.apache.hadoop.io.compress.SnappyCodec</value> 
</property>

My name is len
I love China

I am from China

My                       	1
name                   		1
is                          1
len                         1
I                           2
love                        1 
China                  		2
am                     		1
from                    	1

public class WordCount {
    public WordCount() {
    }
     public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();
        if(otherArgs.length < 2) {
            System.err.println("Usage: wordcount <in> [<in>...] <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(WordCount.TokenizerMapper.class);
        job.setCombinerClass(WordCount.IntSumReducer.class);
        job.setReducerClass(WordCount.IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class); 
        for(int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true)?0:1);
    }
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private static final IntWritable one = new IntWritable(1);
        private Text word = new Text();
        public TokenizerMapper() {
        }
        public void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString()); 
            while(itr.hasMoreTokens()) {
                this.word.set(itr.nextToken());
                context.write(this.word, one);
            }
        }
    }
public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();
        public IntSumReducer() {
        }
        public void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            int sum = 0;
            IntWritable val;
            for(Iterator i$ = values.iterator(); i$.hasNext(); sum += val.get()) {
                val = (IntWritable)i$.next();
            }
            this.result.set(sum);
            context.write(key, this.result);
        }
    }
}

hadoop@Master:/opt/moudle/hadoop$ cd /usr/local/hadoop
hadoop@Master:/opt/moudle/hadoop$ mkdir myapp

hadoop@Master:~$ cd /opt/moudle/hadoop/myapp/
hadoop@Master:/opt/moudle/hadoop/myapp$ ll
总用量 93976
drwxrwxr-x  2 hadoop hadoop     4096 9月  27 15:16 ./
drwxr-xr-x 14 hadoop hadoop     4096 9月  28 20:11 ../
-rw-rw-r--  1 hadoop hadoop 57722832 9月  27 15:16 HDFSExample.jar
-rw-rw-r--  1 hadoop hadoop     5889 9月  20 15:06 MapReduceDemo.jar
-rw-rw-r--  1 hadoop hadoop 38487431 9月  28 20:18 WordCount.jar

hadoop@Master:/opt/moudle/hadoop$ start-all.sh

hadoop@Master:/opt/moudle/hadoop$ hdfs dfs -rm -r /user/hadoop/output

hadoop@Master:/opt/moudle/hadoop$ hdfs dfs -put wordfile1.txt /user/hadoop/input
2021-09-28 20:11:27,930 INFO sasl.SaslDataTransferClient: SASL encryption trust check: localHostTrusted = false, remoteHostTrusted = false
hadoop@Master:/opt/moudle/hadoop$ hdfs dfs -put wordfile2.txt /user/hadoop/input
2021-09-28 20:11:34,434 INFO sasl.SaslDataTransferClient: SASL encryption trust check: localHostTrusted = false, remoteHostTrusted = false

hadoop@Master:/opt/moudle/hadoop$ hdfs dfs -ls /user/hadoop/input
Found 2 items
-rw-r--r--   3 hadoop supergroup         28 2021-09-28 20:11 /user/hadoop/input/wordfile1.txt
-rw-r--r--   3 hadoop supergroup         16 2021-09-28 20:11 /user/hadoop/input/wordfile2.txt

hadoop@Master:~$ cd /opt/moudle/hadoop/myapp/
hadoop@Master:/opt/moudle/hadoop/myapp$ hadoop jar WordCount.jar input output

……//这里省略若干屏幕信息
2021-09-28 20:26:55,683 INFO mapreduce.Job: Running job: job_1632830686414_0001
2021-09-28 20:27:03,779 INFO mapreduce.Job: Job job_1632830686414_0001 running in uber mode : false
2021-09-28 20:27:03,782 INFO mapreduce.Job:  map 0% reduce 0%
2021-09-28 20:27:08,847 INFO mapreduce.Job:  map 100% reduce 0%
2021-09-28 20:27:16,889 INFO mapreduce.Job:  map 100% reduce 100%
2021-09-28 20:27:16,897 INFO mapreduce.Job: Job job_1632830686414_0001 completed successfully
2021-09-28 20:27:16,983 INFO mapreduce.Job: Counters: 53
	File System Counters
		FILE: Number of bytes read=116
		FILE: Number of bytes written=653040
		FILE: Number of read operations=0
		FILE: Number of large read operations=0
		FILE: Number of write operations=0
		HDFS: Number of bytes read=274
		HDFS: Number of bytes written=54
		HDFS: Number of read operations=11
		HDFS: Number of large read operations=0
		HDFS: Number of write operations=2
	Job Counters 
		Launched map tasks=2
		Launched reduce tasks=1
		Data-local map tasks=2
		Total time spent by all maps in occupied slots (ms)=5887
		Total time spent by all reduces in occupied slots (ms)=4688
		Total time spent by all map tasks (ms)=5887
		Total time spent by all reduce tasks (ms)=4688
		Total vcore-milliseconds taken by all map tasks=5887
		Total vcore-milliseconds taken by all reduce tasks=4688
		Total megabyte-milliseconds taken by all map tasks=6028288
		Total megabyte-milliseconds taken by all reduce tasks=4800512
	Map-Reduce Framework
		Map input records=3
		Map output records=11
		Map output bytes=88
		Map output materialized bytes=122
		Input split bytes=230
		Combine input records=11
		Combine output records=11
		Reduce input groups=9
		Reduce shuffle bytes=122
		Reduce input records=11
		Reduce output records=9
		Spilled Records=22
		Shuffled Maps =2
		Failed Shuffles=0
		Merged Map outputs=2
		GC time elapsed (ms)=321
		CPU time spent (ms)=1510
		Physical memory (bytes) snapshot=770977792
		Virtual memory (bytes) snapshot=7914119168
		Total committed heap usage (bytes)=689963008
		Peak Map Physical memory (bytes)=291536896
		Peak Map Virtual memory (bytes)=2637176832
		Peak Reduce Physical memory (bytes)=190496768
		Peak Reduce Virtual memory (bytes)=2641076224
	Shuffle Errors
		BAD_ID=0
		CONNECTION=0
		IO_ERROR=0
		WRONG_LENGTH=0
		WRONG_MAP=0
		WRONG_REDUCE=0
	File Input Format Counters 
		Bytes Read=44
	File Output Format Counters 
		Bytes Written=54

hadoop@Master:/opt/moudle/hadoop/myapp$ hdfs dfs -cat output/*

2020-01-27,Maricopa,Arizona,04013,1,0
2020-01-27,Los Angeles,California,06037,1,0
2020-01-27,Orange,California,06059,1,0
2020-01-27,Cook,Illinois,17031,1,0
2020-01-27,Snohomish,Washington,53061,1,0

    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.1.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.1.3</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.30</version>
        </dependency>
    </dependencies>

log4j.rootLogger=INFO, stdout  
log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
log4j.appender.logfile=org.apache.log4j.FileAppender  
log4j.appender.logfile.File=target/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

/**
 * @author 不温卜火
 * @create 2021-09-30 12:57
 * @MyBlog https://buwenbuhuo.blog.csdn.net
 */
public class CovidCountBean implements Writable {

    private long cases;//确诊病例数
    private long deaths;//死亡病例数

    public CovidCountBean() {
    }

    public CovidCountBean(long cases, long deaths) {
        this.cases = cases;
        this.deaths = deaths;
    }

    public void set(long cases, long deaths) {
        this.cases = cases;
        this.deaths = deaths;
    }

    public long getCases() {
        return cases;
    }

    public void setCases(long cases) {
        this.cases = cases;
    }

    public long getDeaths() {
        return deaths;
    }

    public void setDeaths(long deaths) {
        this.deaths = deaths;
    }

    /**
     *  序列化方法
     */
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(cases);
        out.writeLong(deaths);
    }

    /**
     * 反序列化方法 注意顺序
     */
    @Override
    public void readFields(DataInput in) throws IOException {
        this.cases = in.readLong();
        this.deaths =in.readLong();
    }

    @Override
    public String toString() {
        return  cases +"\t"+ deaths;
    }

}

public class CovidSumMapper extends Mapper<LongWritable, Text, Text, CovidCountBean> {

    Text outKey = new Text();
    CovidCountBean outValue = new CovidCountBean();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] fields = value.toString().split(",");
        //州
        outKey.set(fields[2]);
        //Covid数据 确诊病例 死亡病例
        outValue.set(Long.parseLong(fields[fields.length-2]),Long.parseLong(fields[fields.length-1]));
        //map输出结果
        context.write(outKey,outValue);
    }
}

public class CovidSumReducer extends Reducer<Text, CovidCountBean,Text,CovidCountBean> {

    CovidCountBean outValue = new CovidCountBean();

    @Override
    protected void reduce(Text key, Iterable<CovidCountBean> values, Context context) throws IOException, InterruptedException {
        long totalCases = 0;
        long totalDeaths =0;
        //累加统计
        for (CovidCountBean value : values) {
            totalCases += value.getCases();
            totalDeaths +=value.getDeaths();
        }

        outValue.set(totalCases,totalDeaths);

        context.write(key,outValue);
    }
}

// Mapreduce 自定义对象
public class CovidSumDriver {
    public static void main(String[] args) throws Exception{
        //配置文件对象
        Configuration conf = new Configuration();
        // 创建作业实例
        Job job = Job.getInstance(conf, CovidSumDriver.class.getSimpleName());
        // 设置作业驱动类
        job.setJarByClass(CovidSumDriver.class);

        // 设置作业mapper reducer类
        job.setMapperClass(CovidSumMapper.class);
        job.setReducerClass(CovidSumReducer.class);

        // 设置作业mapper阶段输出key value数据类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(CovidCountBean.class);
        //设置作业reducer阶段输出key value数据类型 也就是程序最终输出数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(CovidCountBean.class);

        // 配置作业的输入数据路径
        FileInputFormat.setInputPaths(job, new Path("E:\\inputCOVID"));
        // 配置作业的输出数据路径
        FileOutputFormat.setOutputPath(job, new Path("E:\\outputCOVID"));


        //判断输出路径是否存在 如果存在删除
        Path outputPath = new Path("E:\\outputCOVID");
        outputPath.getFileSystem(conf).delete(outputPath,true);

        // 提交作业并等待执行完成
        boolean resultFlag = job.waitForCompletion(true);
        //程序退出
        System.exit(resultFlag ? 0 :1);
    }
}

public class CovidCountBean implements WritableComparable<CovidCountBean> {

    private long cases;//确诊病例数
    private long deaths;//死亡病例数

    public CovidCountBean() {
    }

    public CovidCountBean(long cases, long deaths) {
        this.cases = cases;
        this.deaths = deaths;
    }

    public void set(long cases, long deaths) {
        this.cases = cases;
        this.deaths = deaths;
    }

    public long getCases() {
        return cases;
    }

    public void setCases(long cases) {
        this.cases = cases;
    }

    public long getDeaths() {
        return deaths;
    }

    public void setDeaths(long deaths) {
        this.deaths = deaths;
    }

    /**
     *  序列化方法
     */
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(cases);
        out.writeLong(deaths);
    }

    /**
     * 反序列化方法 注意顺序
     */
    @Override
    public void readFields(DataInput in) throws IOException {
        this.cases = in.readLong();
        this.deaths =in.readLong();
    }

    @Override
    public String toString() {
        return  cases +"\t"+ deaths;
    }

    /**
     * 排序比较器  本业务中根据确诊案例数倒序排序
     */
    @Override
    public int compareTo(CovidCountBean o) {
        return this.cases - o.getCases()> 0 ? -1:(this.cases - o.getCases() < 0 ? 1 : 0);
    }
}

public class CovidSortSumMapper extends Mapper<LongWritable, Text, CovidCountBean,Text> {

    CovidCountBean outKey = new CovidCountBean();
    Text outValue = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] fields = value.toString().split("\t");
        outKey.set(Long.parseLong(fields[1]),Long.parseLong(fields[2]));
        outValue.set(fields[0]);
        context.write(outKey,outValue);
    }
}

public class CovidSortSumReducer extends Reducer<CovidCountBean, Text,Text,CovidCountBean> {
    @Override
    protected void reduce(CovidCountBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        Text outKey = values.iterator().next();
        context.write(outKey,key);
    }
}

public class CovidSortSumDriver {
    public static void main(String[] args) throws Exception{
        //配置文件对象
        Configuration conf = new Configuration();
        // 创建作业实例
        Job job = Job.getInstance(conf, CovidSortSumDriver.class.getSimpleName());
        // 设置作业驱动类
        job.setJarByClass(CovidSortSumDriver.class);

        // 设置作业mapper reducer类
        job.setMapperClass(CovidSortSumMapper.class);
        job.setReducerClass(CovidSortSumReducer.class);

        // 设置作业mapper阶段输出key value数据类型
        job.setMapOutputKeyClass(CovidCountBean.class);
        job.setMapOutputValueClass(Text.class);
        //设置作业reducer阶段输出key value数据类型 也就是程序最终输出数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(CovidCountBean.class);

        // 配置作业的输入数据路径
        FileInputFormat.setInputPaths(job, new Path("E:\\outputCOVID"));
        // 配置作业的输出数据路径
        FileOutputFormat.setOutputPath(job, new Path("E:\\outputCOVID1"));


        //判断输出路径是否存在 如果存在删除
        Path outputPath = new Path("E:\\outputCOVID1");
        outputPath.getFileSystem(conf).delete(outputPath,true);

        // 提交作业并等待执行完成
        boolean resultFlag = job.waitForCompletion(true);
        //程序退出
        System.exit(resultFlag ? 0 :1);
    }
}

public class StatePartitioner extends Partitioner<Text, Text> {

    //模拟美国各州数据字典 实际中可以从redis中快速查询  如果数据不大也可以使用数据集合保存
    public static HashMap<String, Integer> stateMap = new HashMap<String, Integer>();

    static{
        stateMap.put("Alabama", 0);
        stateMap.put("Arkansas", 1);
        stateMap.put("California", 2);
        stateMap.put("Florida", 3);
        stateMap.put("Indiana", 4);
    }

    @Override
    public int getPartition(Text key, Text value, int numPartitions) {
        Integer code = stateMap.get(key.toString());

        if (code != null) {
            return code;
        }

        return 5;
    }
}

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

    Text outKey = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] splits = value.toString().split(",");
        //以州作为输出的key
        outKey.set(splits[2]);
        context.write(outKey,value);
    }
}

public class CovidPartitionReducer extends Reducer<Text,Text,Text, NullWritable> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            context.write(value,NullWritable.get());
        }
    }
}

public class CovidPartitionDriver {
    public static void main(String[] args) throws Exception{
        //配置文件对象
        Configuration conf = new Configuration();
        // 创建作业实例
        Job job = Job.getInstance(conf, CovidPartitionDriver.class.getSimpleName());
        // 设置作业驱动类
        job.setJarByClass(CovidPartitionDriver.class);

        // 设置作业mapper reducer类
        job.setMapperClass(CovidPartitionMapper.class);
        job.setReducerClass(CovidPartitionReducer.class);

        // 设置作业mapper阶段输出key value数据类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        //设置作业reducer阶段输出key value数据类型 也就是程序最终输出数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //todo 设置reducetask个数  和自定义分区器
        job.setNumReduceTasks(8);

        job.setPartitionerClass(StatePartitioner.class);

        // 配置作业的输入数据路径
        FileInputFormat.setInputPaths(job, new Path("E:\\inputCOVID"));
        // 配置作业的输出数据路径
        FileOutputFormat.setOutputPath(job, new Path("E:\\outputCOVID2"));


        //判断输出路径是否存在 如果存在删除
        Path outputPath = new Path("E:\\outputCOVID2");
        outputPath.getFileSystem(conf).delete(outputPath,true);

        // 提交作业并等待执行完成
        boolean resultFlag = job.waitForCompletion(true);
        //程序退出
        System.exit(resultFlag ? 0 :1);
    }
}

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

    private IntWritable outV = 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();
        }

        //封装outKV
        outV.set(sum);

        //写出outKV
        context.write(key,outV);
    }
}

// 指定需要使用Combiner，以及用哪个类作为Combiner的逻辑
job.setCombinerClass(WordCountReducer.class);

public class CovidBean implements WritableComparable<CovidBean> {

    private String state;//州
    private String county;//县
    private long cases;//确诊病例

    public CovidBean() {
    }

    public CovidBean(String state, String county, long cases) {
        this.state = state;
        this.county = county;
        this.cases = cases;
    }

    public void set (String state, String county, long cases) {
        this.state = state;
        this.county = county;
        this.cases = cases;
    }

    public String getState() {
        return state;
    }

    public void setState(String state) {
        this.state = state;
    }

    public String getCounty() {
        return county;
    }

    public void setCounty(String county) {
        this.county = county;
    }

    public long getCases() {
        return cases;
    }

    public void setCases(long cases) {
        this.cases = cases;
    }

    @Override
    public String toString() {
        return "CovidBean{" +
                "state='" + state + '\'' +
                ", county='" + county + '\'' +
                ", cases=" + cases +
                '}';
    }

    //todo 排序规则 根据州state正序进行排序 如果州相同 则根据确诊数量cases倒序排序
    @Override
    public int compareTo(CovidBean o) {

        int result ;

        int i = state.compareTo(o.getState());

        if ( i > 0) {
            result =1;
        } else if (i <0 ) {
            result = -1;
        } else {
            // 确诊病例数倒序排序
            result = cases > o.getCases() ? -1 : 1;
        }

        return result;

    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(state);
        out.writeUTF(county);
        out.writeLong(cases);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.state =in.readUTF();
        this.county =in.readUTF();
        this.cases =in.readLong();
    }
}

public class CovidTop1Mapper extends Mapper<LongWritable, Text, CovidBean, NullWritable> {

    CovidBean outKey = new CovidBean();
    NullWritable outValue = NullWritable.get();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] fields = value.toString().split(",");
        //封装数据： 州 县 确诊病例
        outKey.set(fields[2],fields[1],Long.parseLong(fields[4]));
        context.write(outKey,outValue);
    }
}

public class CovidGroupingComparator extends WritableComparator {

    protected CovidGroupingComparator(){
        super(CovidBean.class,true);
    }

    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        CovidBean aBean = (CovidBean) a;
        CovidBean bBean = (CovidBean) b;

        return aBean.getState().compareTo(bBean.getState());
    }
}

public class CovidTop1Reducer extends Reducer<CovidBean, NullWritable,CovidBean,NullWritable> {
    @Override
    protected void reduce(CovidBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        //不遍历迭代器，此时key就是分组中的第一个key 也就是该州确诊病例数最多的县对应的数据
        context.write(key,NullWritable.get());
    }
}

public class CovidTop1Driver {
    public static void main(String[] args) throws Exception{
        //配置文件对象
        Configuration conf = new Configuration();
        // 创建作业实例
        Job job = Job.getInstance(conf, CovidTop1Driver.class.getSimpleName());
        // 设置作业驱动类
        job.setJarByClass(CovidTop1Driver.class);

        // 设置作业mapper reducer类
        job.setMapperClass(CovidTop1Mapper.class);
        job.setReducerClass(CovidTop1Reducer.class);

        // 设置作业mapper阶段输出key value数据类型
        job.setMapOutputKeyClass(CovidBean.class);
        job.setMapOutputValueClass(NullWritable.class);
        //设置作业reducer阶段输出key value数据类型 也就是程序最终输出数据类型
        job.setOutputKeyClass(CovidBean.class);
        job.setOutputValueClass(NullWritable.class);

        //todo 设置自定义分组
        job.setGroupingComparatorClass(CovidGroupingComparator.class);

        // 配置作业的输入数据路径
        FileInputFormat.setInputPaths(job, new Path("E:\\inputCOVID"));
        // 配置作业的输出数据路径
        FileOutputFormat.setOutputPath(job, new Path("E:\\outputCOVID3"));


        //判断输出路径是否存在 如果存在删除
        Path outputPath = new Path("E:\\outputCOVID3");
        outputPath.getFileSystem(conf).delete(outputPath,true);

        // 提交作业并等待执行完成
        boolean resultFlag = job.waitForCompletion(true);
        //程序退出
        System.exit(resultFlag ? 0 :1);
    }
}

public class StatePartitioner extends Partitioner<Text, Text> {

    //模拟美国各州数据字典 实际中可以从redis中快速查询  如果数据不大也可以使用数据集合保存
    public static HashMap<String, Integer> stateMap = new HashMap<String, Integer>();

    static{
        stateMap.put("Alabama", 0);
        stateMap.put("Arkansas", 1);
        stateMap.put("California", 2);
        stateMap.put("Florida", 3);
        stateMap.put("Indiana", 4);
    }

    @Override
    public int getPartition(Text key, Text value, int numPartitions) {
        Integer code = stateMap.get(key.toString());

        if (code != null) {
            return code;
        }

        return 5;
    }
}

public class CovidTopNMapper extends Mapper<LongWritable, Text, CovidBean,LongWritable> {
    CovidBean outKey = new CovidBean();
    LongWritable outValue = new LongWritable();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] fields = value.toString().split(",");
        //封装数据： 州 县 确诊病例
        outKey.set(fields[2],fields[1],Long.parseLong(fields[4]));
        outValue.set(Long.parseLong(fields[4]));

        context.write(outKey,outValue);
    }

public class CovidTopNReducer extends Reducer<CovidBean, LongWritable,CovidBean,LongWritable> {
    @Override
    protected void reduce(CovidBean key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        int num =0;
        for (LongWritable value : values) {
            if(num < 10 ){ //输出每个州最多的前10个
                context.write(key,value);
                num++;
            }else{
                return;
            }
        }
    }
}

public class CovidTopNDriver {
    public static void main(String[] args) throws Exception{
        //配置文件对象
        Configuration conf = new Configuration();
        // 创建作业实例
        Job job = Job.getInstance(conf, CovidTopNDriver.class.getSimpleName());
        // 设置作业驱动类
        job.setJarByClass(CovidTopNDriver.class);

        // 设置作业mapper reducer类
        job.setMapperClass(CovidTopNMapper.class);
        job.setReducerClass(CovidTopNReducer.class);

        // 设置作业mapper阶段输出key value数据类型
        job.setMapOutputKeyClass(CovidBean.class);
        job.setMapOutputValueClass(LongWritable.class);
        //设置作业reducer阶段输出key value数据类型 也就是程序最终输出数据类型
        job.setOutputKeyClass(CovidBean.class);
        job.setOutputValueClass(LongWritable.class);

        //todo 设置自定义分组
        job.setGroupingComparatorClass(CovidGroupingComparator.class);

        // 配置作业的输入数据路径
        FileInputFormat.setInputPaths(job, new Path("E:\\inputCOVID"));
        // 配置作业的输出数据路径
        FileOutputFormat.setOutputPath(job, new Path("E:\\outputCOVID4"));


        //判断输出路径是否存在 如果存在删除
        Path outputPath = new Path("E:\\outputCOVID4");
        outputPath.getFileSystem(conf).delete(outputPath,true);

        // 提交作业并等待执行完成
        boolean resultFlag = job.waitForCompletion(true);
        //程序退出
        System.exit(resultFlag ? 0 :1);
    }
}