package mobvista.prd.datasource.country.interest.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.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.Reducer; 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 Administrator on 2017/5/12 0012. */ public class CountryInterestMR { public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException { Configuration conf = new Configuration(); conf.set("mapreduce.map.speculative", "true"); conf.set("mapreduce.reduce.speculative", "true"); conf.set("mapreduce.job.reduce.slowstart.completedmaps", "1.0"); conf.set("mapreduce.job.speculative.slowtaskthreshold", "1.0"); conf.set("mapreduce.reduce.memory.mb", "2250"); String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); Job job = Job.getInstance(conf, "CountryInterestMR"); FileOutputFormat.setCompressOutput(job, true); FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); job.setNumReduceTasks(Integer.parseInt(otherArgs[3])); job.setJarByClass(CountryInterestMR.class); job.setMapperClass(CountryInterestMapper.class); job.setReducerClass(CountryInterestReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileInputFormat.addInputPath(job, new Path(otherArgs[1])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[2])); System.exit(job.waitForCompletion(true) ? 0 : 1); } public static class CountryInterestMapper extends Mapper<LongWritable ,Text, Text, Text> { Text outKey = new Text(); Text outValue = new Text(); public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { try { String[] arr = MRUtils.SPLITTER.split(value.toString(), -1); String inputFile = context.getConfiguration().get("map.input.file"); if (inputFile.contains("etl_dsp_request_daily")) { //dsp outKey.set(MRUtils.JOINER.join(arr[0], arr[1]));//device_id,device_type outValue.set(arr[3] + "\t" + "");//国家 context.write(outKey, outValue); } else if (inputFile.contains("dm_interest_tag")) { //dm_interest_tag outKey.set(MRUtils.JOINER.join(arr[0], arr[1]));//device_id,device_type outValue.set(arr[3]);//兴趣 context.write(outKey, outValue); } else if (inputFile.contains("etl_adn_sdk_request_daily")) { // M系统 outKey.set(MRUtils.JOINER.join(arr[0], arr[1]));//device_id,device_type outValue.set(arr[2] + "\t" + "");//国家 context.write(outKey, outValue); } } catch (Exception e) { return; } } } public static class CountryInterestReducer extends Reducer<Text, Text, Text, Text> { Text outKey = new Text(); Text outValue = new Text(); public void reduce (Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { String country = ""; String interest = ""; for (Text val : values) { if (val.toString().contains("\t")) { country = val.toString().split("\t",-1)[0]; } else { interest = val.toString(); } } if (country.equals("")) { return; } if (interest.equals("")) { return; } outValue.set(MRUtils.JOINER.join(country,interest)); context.write(key, outValue); } } }