电影票房之数据清洗（MapReduce）

第1关：数据清洗

任务描述

基于EduCoder平台提供的初始数据集（数据集存放在 /data/workspace/myshixun/data/movies.csv 中），按照下面的要求，完成电影票房数据的清洗工作。

编程要求

1. 去掉字段“上映天数”中带有“零点场”、“点映”、“展映”和“重映”的电影数据；

2. 以字段“上映天数”和“当前日期”为依据，在尾列添加一个“上映日期”（releaseDate）的字段，该字段值为“当前日期”减去“上映天数”+1（格式为：2020-10-13）。例如：若字段“上映天数”的值为“上映2天”，字段“当前日期”为“2020-10-10”，则字段“上映日期”的值为“2020-10-09”。如果字段“上映天数”为空，则字段“上映日期”的值设为“往期电影”。注意：若字段“上映天数”的值为“上映首日”，则字段“上映日期”的值应设为“当前日期”的值；

3. 对字段“当日综合票房”和字段“当前总票房”的数据值进行处理，单位量级统一以“万”展示数据，若原数据中有“万”等表示量级的汉字，需去掉量级汉字。如：“1.5亿”需转换为“15000”，“162.4万”转换为“162.4”，转换时需注意精度缺失问题，当我们使用 double、float 等类型做算术运算时，例如：1.14 亿转换为万时，结果会出现 11399.999999999998 万，这里我们可以使用 BigDecimal 类来处理这类精度缺失问题，字段“当日综合票房”和字段“当前总票房”最后保留两位小数；

4. 清洗完的数据集存储到 /root/files 目录下，分隔方式为 \t。

注意：不要改变字段原本的顺序。

数据集说明

本数据集是电影票房数据，包含十个字段的信息，数据集的字段含义说明如下：

字段名	说明
movie_name	电影名
boxoffice	当日综合票房
box_rate	票房占比
sessions	排片场次
show_count_rate	排片占比
avg_number	场均人次
attendance	上座率
total_boxoffice	当前总票房
movie_days	上映天数
current_time	当前日期

部分数据展示：

盛世秧歌    7.06万    <0.1%    107    <0.1%    22    47.3%    1757.1万    展映    2019-10-16
决胜时刻    70.83万    0.6%    561    0.1%    34    21.6%    1.14亿    上映25天    2019-10-14
天池水怪    5.43万    <0.1%    55    <0.1%    30    --    75.1万        2020-10-06

清洗后数据展示：

决胜时刻    70.83    0.6%    561    0.1%    34    21.6%    11400.0    上映25天    2019-10-14    2019-09-20
天池水怪    5.43    <0.1%    55    <0.1%    30    --    75.1        2020-10-06    往期电影

注意事项

• 任务完成后请点击"评测"按钮，验证电影票房数据是否成功清洗，程序未通过的情况下，可以点击测试集查看具体问题；

• 本任务设置了多个测试集，通过其中任一测试集将可获得对应分数。

考核点说明

1. 清洗数据后共多少行；
2. 字段“上映日期”添加成功；
3. 字段“当日综合票房”和字段“当前总票房”的单位处理成功。

---

开始你的任务吧，祝你成功！

MovieMap.java

package movie.mapreduce;
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;
import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.GregorianCalendar;
import java.text.DecimalFormat;

public class MovieMap extends Mapper<LongWritable, Text,Text,NullWritable> {
    /********** Begin **********/
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] values=value.toString().split(",");
        boolean flag=true;
        if (values[8].contains("零点场")||values[8].contains("点映")||values[8].contains("展映")||values[8].contains("重映")){
            flag=false;
        }
        if (flag){
            SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd");
            String releaseDate="";
            if (values[8].trim()==null||values[8].trim().equals("")){
                releaseDate="往期电影";
            }else if (values[8].trim().equals("上映首日")){
                releaseDate=values[9];
            }else{
                int movie_days=Integer.parseInt(values[8].replaceAll("[^0-9]",""))-1;
                Date current_time= null;
                try {
                    current_time = sdf.parse(values[9]);
                    //日历类，用于日期计算
                    Calendar c = new GregorianCalendar();
                    c.setTime(current_time);
                    c.add(Calendar.DATE,-movie_days);
                    releaseDate=sdf.format(c.getTime());
                } catch (ParseException e) {
                    e.printStackTrace();
                }
            }
            values[1]=boxofficeParse(values[1]);
            values[7]=boxofficeParse(values[7]);
            String result="";
            for (int i=0;i<values.length;i++){
                if (i==values.length-1){
                    result=result+values[i];
                }else{
                    result=result+values[i]+"\t";
                }
            }
            result=result+"\t"+releaseDate;
            context.write(new Text(result),NullWritable.get());
        }
    }
    static String boxofficeParse(String s){
        String boxoffice = "";
        if (s.contains("万")) {
            boxoffice = s.split("万")[0];
        } else if (s.contains("亿")) {
            BigDecimal b1 = new BigDecimal(s.split("亿")[0]);
            BigDecimal b2 = new BigDecimal(10000);
            //结果保留两位小数
            DecimalFormat df = new DecimalFormat("#0.00");
            //b1*b2，取double类型，结果保留2位小数
            boxoffice = df.format(b1.multiply(b2).doubleValue());
        }
        return boxoffice;
    }
    /********** Begin **********/
}

MovieReduce.java

package movie.mapreduce;

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

import java.io.IOException;

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

MovieJob.java

package movie.mapreduce;


import org.apache.hadoop.conf.Configuration;
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.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class MovieJob {
    public static void main(String[] args) throws Exception {
        /********** Begin **********/
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(MovieJob.class);
        job.setMapperClass(MovieMap.class);
        job.setReducerClass(MovieReduce.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        String outputpath = "/root/files";
        Path path= new Path(outputpath);
        FileSystem fileSystem =path.getFileSystem(conf);
        if (fileSystem.exists(path)){
            fileSystem.delete(path,true);
        }
        FileInputFormat.addInputPath(job, new Path("/data/workspace/myshixun/data/movies.csv"));
        FileOutputFormat.setOutputPath(job,path);
        job.waitForCompletion(true);
       
       
        /********** End **********/
    }
}