数据清洗案例实操

LogDriver

public class LogDriver {

    public static void main(String[] args) throws Exception {

        // 0 根据自己电脑路径重新配置

        args = new String[]{"e:/input/inputlog", "e:/output1"};

        // 1 获取job信息

        Configuration configuration = new Configuration();

        Job job = Job.getInstance(configuration);

        // 2 设置加载jar包路径

        job.setJarByClass(LogDriver.class);

        // 3 关联map

        job.setMapperClass(LogMapper.class);

        // 4 设置最终输出数据类型

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(NullWritable.class);

        // 5 Map端Join（的逻辑不需要Reducer阶段，设置reducerTask数量为0

        job.setNumReduceTasks(0);

        // 6 设置输入输出路径

        FileInputFormat.setInputPaths(job, new Path(args[0]));

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

        // 7 提交

        job.waitForCompletion(true);

    }

}

LogMapper

public class LogMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    @Override

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        // 1 获取一行

        String line = value.toString();

        // 2 解析数据

        boolean result = parseLog(line, context);

        if(!result){

            return;

        }

        // 3 解析通过 写出去

        context.write(value, NullWritable.get());

    }

    private boolean parseLog(String line, Context context){

        String[] fields = line.split(" ");

        if (fields.length > 11 ){

            context.getCounter("map","true").increment(1);

            return true;

        }else{

            context.getCounter("map","false").increment(1);

            return false;

        }

    }

}