题目和要求

题目

• 2020年新冠肺炎对我国社会各方面影响巨大，大数据技术在抗击疫情过程中发挥了巨大作用，尤其在新增、确认等相关病例数据的采集及统计上应用颇广，下面有一份数据是今年1月20-4月29日的全国各省市及国外的疫情数据，请你按照要求使用MapReduce程序完成相关数据预处理。
• 数据地址

第一小题

• 数据转换：请将数据中日期字段格式，替换成日期格式为xxxx年xx月xx日
• 文件中的日期格式是
  

Map阶段

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

import java.io.IOException;

public class map extends Mapper<LongWritable, Text,Text, NullWritable> {
    Text k = new Text();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //加-1是因为不确定最后是否不为空
        //切割一行数据
        String [] text = value.toString().split(",",-1);
        //拼接数据
        String date = "2020年"+text[0];
        k.set(date+","+text[1]+text[2]+","+text[3]+","+text[4]+","+text[5]+","+text[6]+","+text[7]);
        context.write(k,NullWritable.get());
    }
}

Reduce阶段

• 不需要做更改直接输出即可

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

import java.io.IOException;

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

Driver阶段

• 设置输入和输出路径

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

import java.io.File;

public class Driver {
    public static void main(String[] args) throws Exception{
        //判断输出路径是否存在，存在的话就删除
        File file = new File("D:\\MPTest\\dateoutput");
        if (file.exists()){
            delFile(file);
            driver();
        }else {
            driver();
        }
    }
    //删除输出路径下的一切文件夹和文件
    public static void delFile(File file) {
        File[] files = file.listFiles();
        if (files != null && files.length != 0) {
            for (int i = 0;i<files.length;i++) {
                delFile(files[i]);
            }
        }
        file.delete();
    }
    public  static void driver() throws Exception{
        Job job = Job.getInstance(new Configuration());

        //设置Map、Reduce和主类
        job.setMapperClass(map.class);
        job.setReducerClass(Red.class);
        job.setJarByClass(Driver.class);

        //分别设置Map和Reduce阶段输出的KV类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        //设置输入路径
        FileInputFormat.setInputPaths(job, "D:\\MPTest\\dateinput\\data.csv");

        //设置输出路径
        FileOutputFormat.setOutputPath(job, new Path("D:\\MPTest\\dateoutput"));

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

第二小题

• 数据清洗：以下规则同时进行

3. 规则1 从上述小题中，截取前5个字段。
4. 规则2 过滤出省份为湖北省的数据。
5. 规则3 对5个字段去重,生成新的数据。

简要分析

• 截取前5个字段，要对题1的数据再次进行分割
• 过滤出湖北省的数据，要重写分区方法
• 对字段进行去重操作，重写他的排序，以每个对象中的全部字符排序，在Reduce阶段，只输出一次即可完成去重操作

重写一个类来存储数据

import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Objects;

public class data implements WritableComparable<data> {
    //定义五个变量来存储五个字段(不要在意取名！！！！！)
    private String date;
    private String localtion;
    private String a;
    private String b;
    private String c;

    //编写set方法，在Map阶段可以快捷的赋值
    public void set(String date, String localtion, String a, String b, String c) {
        this.date = date;
        this.localtion = localtion;
        this.a = a;
        this.b = b;
        this.c = c;
    }

    //重写比较方法
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        data data = (data) o;
        return a == data.a &&
                b == data.b &&
                c == data.c &&
                Objects.equals(date, data.date) &&
                Objects.equals(localtion, data.localtion);
    }

    @Override
    public int hashCode() {
        return Objects.hash(date, localtion, a, b, c);
    }

    public String getDate() {
        return date;
    }

    public void setDate(String date) {
        this.date = date;
    }

    public String getLocaltion() {
        return localtion;
    }

    public void setLocaltion(String localtion) {
        this.localtion = localtion;
    }

    public String getA() {
        return a;
    }

    public void setA(String a) {
        this.a = a;
    }

    public String getB() {
        return b;
    }

    public void setB(String b) {
        this.b = b;
    }

    public String getC() {
        return c;
    }

    public void setC(String c) {
        this.c = c;
    }

    //重写排序方法，五个字符
    @Override
    public int compareTo(data o) {
        int i = o.date.compareTo(this.date);
        int j = o.localtion.compareTo(this.localtion);
        int n = o.a.compareTo(this.a);
        int m = o.b.compareTo(this.b);
        int l = o.c.compareTo(this.c);
        if (i == 0) {
            if (j == 0) {
                if (n == 0) {
                    if (m == 0) {
                        return l;
                    } else {
                        return m;
                    }
                } else {
                    return n;
                }
            } else {
                return j;
            }
        } else {
            return i;
        }
    }

    //序列化
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(date);
        dataOutput.writeUTF(localtion);
        dataOutput.writeUTF(a);
        dataOutput.writeUTF(b);
        dataOutput.writeUTF(c);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.date = dataInput.readUTF();
        this.localtion = dataInput.readUTF();
        this.a = dataInput.readUTF();
        this.b = dataInput.readUTF();
        this.c = dataInput.readUTF();
    }

    //重写toString方法，好输出数据
    @Override
    public String toString() {
        return date + ',' + localtion + ',' + a + ',' + b + ',' + c + ',';
    }
}

Map阶段

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

import java.io.IOException;

public class Map extends Mapper<LongWritable, Text,data, NullWritable> {
    data data = new data();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //切割每一行数据
        String [] datas = value.toString().split(",",-1);
        //将切割好的数据放到data类里
        data.set(datas[0],datas[1],datas[2],datas[3],datas[4]);
        context.write(data,NullWritable.get());
    }
}

重写分区方法

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

public class dataPartition extends Partitioner<data, NullWritable> {
    @Override
    public int getPartition(data data, NullWritable nullWritable, int i) {
        //比较开头是否是湖北，是的话，符合条件，筛选出来
        if (data.getLocaltion().startsWith("湖北")){
            return 0;
        }
        return 1;
    }
}

Reduce阶段

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

import java.io.IOException;

public class Red extends Reducer<data, NullWritable,data,NullWritable> {
    @Override
    protected void reduce(data key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        //只输出一次，每一组的数据是重复的，只输出一次来达到去重的效果
        context.write(key,NullWritable.get());
    }
}

Driver阶段

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

import java.io.File;

public class Dri {
    public static void main(String[] args) throws Exception {
        File file = new File("D:\\MPTest\\dataoutput");
        if (file.exists()){
            delFile(file);
            driver();
        }else {
            driver();
        }
    }
    public static void delFile(File file) {
        File[] files = file.listFiles();
        if (files != null && files.length != 0) {
            for (int i = 0;i<files.length;i++) {
                delFile(files[i]);
            }
        }
        file.delete();
    }
    public  static void driver() throws Exception{
        Job job = Job.getInstance(new Configuration());

        job.setMapperClass(Map.class);
        job.setReducerClass(Red.class);
        job.setJarByClass(Dri.class);

        job.setMapOutputKeyClass(data.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setOutputKeyClass(data.class);
        job.setMapOutputValueClass(NullWritable.class);

        //设置分区为你自己设定的，并且设置ReduceTask的数量，需要和你的分区数一样
        job.setPartitionerClass(dataPartition.class);
        job.setNumReduceTasks(2);

        FileInputFormat.setInputPaths(job, "D:\\MPTest\\dateoutput");
        FileOutputFormat.setOutputPath(job, new Path("D:\\MPTest\\dataoutput"));

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