MapReduce实战

public class DataCleanMap extends Mapper<LongWritable, Text, Text, Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String line = value.toString();
        JSONObject jsonObject = JSON.parseObject(line);
        String id = jsonObject.getString("uid");
        int gold = jsonObject.getIntValue("gold");
        int watchnumpv = jsonObject.getIntValue("watchnumpv");
        int follower = jsonObject.getIntValue("follower");
        int length = jsonObject.getIntValue("length");
        if (gold >= 0 && watchnumpv >= 0 && follower >= 0 && length >= 0) {
            // 组装k2, v2
            Text k2 = new Text(id);
            Text v2 = new Text(gold + "\t" + watchnumpv + "\t" + follower + "\t" + length);
            context.write(k2, v2);
        }
    }
}

public class DataCleanJob {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        if (args.length != 2) {
            System.exit(100);
        }
        Configuration conf = new Configuration();
        // 创建一个Job
        Job job = Job.getInstance(conf, "DataClean");
        // 设置Job运行的类
        job.setJarByClass(DataCleanJob.class);
        // 设置Job的Mapper类
        job.setMapperClass(DataCleanMap.class);
        // 设置Job的Reduce类
        //job.setReducerClass(WordCountReducer.class);
        // 设置文本输入类型
        job.setInputFormatClass(TextInputFormat.class);
        // 设置文本输出类型
        job.setOutputFormatClass(TextOutputFormat.class);
        // 设置文件输入路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        // 设置文件输出路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // 设置输出文件key, value的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        // 警用reduce
        job.setNumReduceTasks(0);
        // 启动任务
        job.waitForCompletion(true);
    }
}