package mobvista.dmp.datasource.age.mapreduce; import com.google.common.collect.Sets; 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.Text; import org.apache.hadoop.io.compress.GzipCodec; 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; import java.util.Set; /** * Created by liushaui on 2017/2/16 0016. */ public class GetGaAgeMR { 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("now", otherArgs[2]); Job job = Job.getInstance(conf, "GetGaAgeMR"); FileOutputFormat.setCompressOutput(job, true); FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); job.setJarByClass(GetGaAgeMR.class); job.setMapperClass(GetGaGenderMapper.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.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 GetGaGenderMapper extends Mapper<LongWritable, Text, Text, Text> { Text outKey = new Text(); Text outValue = new Text(); int now; Set<String> matchingSet = Sets.newHashSet("", "0", "1970"); public void setup(Context context) throws IOException, InterruptedException { now = Integer.parseInt(context.getConfiguration().get("now")); } public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String[] fields = MRUtils.SPLITTER.split(line, -1); if (fields.length != 33) { return; } String deviceId = fields[0]; String[] tmpDeviceId = Util.lineSplit.split(deviceId, -1); if (tmpDeviceId.length != 5) { return; } if (Util.match.matcher(deviceId).matches()) { return; } if (matchingSet.contains(fields[27])) { return; } int age = now - Integer.parseInt(fields[27]); if (age >= 100 || age <= 0) { return; } int label = Util.calcLabel(age); outKey.set(fields[0]); outValue.set(MRUtils.JOINER.join("A", label, "ga", fields[1])); context.write(outKey, outValue); } } }