package mobvista.prd.datasource.source.mapreduce; import mobvista.dmp.util.MRUtils; import mobvista.prd.datasource.table.MergeAppIDMR; 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.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/18 0018. * desc :dsp 3天设备去重,保留device_id,device_type,国家 */ public class MergeDspCountryMR { public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException { Configuration conf = new Configuration(); String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); Job job = Job.getInstance(conf, "dsp merge"); job.setJarByClass(MergeAppIDMR.class); FileOutputFormat.setCompressOutput(job, true); FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); job.setMapperClass(MergeDspCountryMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); job.setReducerClass(MergeDspCountryReducer.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 MergeDspCountryMapper extends Mapper<LongWritable, Text, Text, NullWritable> { Text outKey = new Text(); Text outValue = new Text(); public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split("\t", -1); outKey.set(MRUtils.JOINER.join(fields[0], fields[1], "dsp系统", fields[3])); //device_id,device_type,country context.write(outKey, NullWritable.get()); } } public static class MergeDspCountryReducer extends Reducer<Text, Text, Text, NullWritable> { Text outKey = new Text(); public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { context.write(key, NullWritable.get()); } } }