package mobvista.dmp.datasource.age.mapreduce; import mobvista.dmp.util.MRUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; import java.io.IOException; /** * Created by liushaui on 2017/2/17 0017. */ public class CalcPackageDictMR { public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException { Configuration conf = new Configuration(); conf.set("mapreduce.map.speculative", "true"); conf.set("mapreduce.reduce.speculative", "true"); conf.set("mapreduce.task.io.sort.mb", "500"); conf.set("mapreduce.reduce.java.opts", "-Xmx1536m"); conf.set("mapreduce.reduce.memory.mb", "2048"); conf.set("mapreduce.reduce.shuffle.parallelcopies", "50"); String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); conf.set("low", otherArgs[2]); conf.set("high", otherArgs[3]); conf.set("unbelievable", otherArgs[4]); Job job = Job.getInstance(conf, "CalcPackageDictMR"); job.setJarByClass(CalcPackageDictMR.class); job.setMapperClass(CalcPackageDictMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } public static class CalcPackageDictMapper extends Mapper<LongWritable, Text, Text, NullWritable> { Text outKey = new Text(); int low; int high; float unbelievable; public void setup(Context context) throws IOException, InterruptedException { low = Integer.parseInt(context.getConfiguration().get("low")); high = Integer.parseInt(context.getConfiguration().get("high")); unbelievable = Float.parseFloat(context.getConfiguration().get("unbelievable")); } public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String[] fields = MRUtils.SPLITTER.split(line, -1); String tag = ""; if (fields[4].equals("0.0") || Integer.parseInt(fields[2]) < low) { return; } if (Float.parseFloat(fields[3]) >= unbelievable) { tag = "unbelievable"; outKey.set(MRUtils.JOINER.join(fields[0], fields[4], tag)); context.write(outKey, NullWritable.get()); } if (Integer.parseInt(fields[2]) >= high) { tag = "confirm"; } else { tag = "calc"; } outKey.set(MRUtils.JOINER.join(fields[0], fields[4], tag)); context.write(outKey, NullWritable.get()); } } }