DmpDeviceTotalMR.java

package mobvista.dmp.device.mapreduce;

import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.map.WrappedMapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer;
import org.apache.hadoop.mapreduce.task.ReduceContextImpl;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;

import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.regex.Pattern;

/**
 * author: houying
 * date : 16-10-8
 * desc :
 */
public class DmpDeviceTotalMR extends Configured implements Tool {
    private static Pattern splitPtn = Pattern.compile("\t");

    public static class DmpDeviceMapper extends Mapper<LongWritable, Text, Text, Text> {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] array = splitPtn.split(line);
            int length = array.length;
            if (length == 13 || length == 34 || length == 45) { //adn || ga || dmp
                context.write(new Text(array[0]), value);
            }
        }
    }

    public static class DmpDeviceReducer extends Reducer<Text, Text, Text, NullWritable> {
        private static Joiner joiner = Joiner.on("\t").useForNull("");
        private DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss");

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            String id = key.toString();
            String[] outputArray = null;
            String[] adnArray = null;
            String[] gaArray = null;
            for (Text value : values) {
                String[] array = splitPtn.split(value.toString());
                if ((adnArray == null || ((adnArray[1].equals("androidid") || adnArray[1].equals("android_id")) && array[1].equals("gaid")))
                        && array.length == 13) { // adn
                    adnArray = array;
                } else if (gaArray == null && array.length == 34) { //ga
                    gaArray = array;
                } else if (outputArray == null && array.length == 45) {
                    outputArray = array;
                }
            }
            if (outputArray == null) {
                outputArray = new String[45];
                Arrays.fill(outputArray, "");
            }
            outputArray[0] = id;
            if (gaArray != null) {
                fillInGa(gaArray, outputArray);
            }
            if (adnArray != null) {
                fillInAdn(adnArray, outputArray);
            }
            for (int i = 2; i < outputArray.length; i++) {
                if (outputArray[i].equals("NULL")) {
                    outputArray[i] = "";
                }
            }
            context.write(new Text(joiner.join(outputArray)), NullWritable.get());
        }

        /**
         * 向outputArray中填充adn数据字段
         *
         * @param array       adn数据字段
         * @param outputArray 输出数组
         */
        private void fillInAdn(String[] array, String[] outputArray) {
            outputArray[1] = array[1]; //device_type
            outputArray[2] = array[3]; //platform
            outputArray[3] = array[4]; //os version
            outputArray[4] = array[6]; //device model
            outputArray[5] = array[7]; //device brand
            outputArray[6] = array[9]; //country code
            outputArray[7] = array[10];//language
            outputArray[8] = array[2]; //mac
            outputArray[43] = array[11]; //adn_app_click_list
            outputArray[44] = array[12]; //update time
        }

        /**
         * 向outputArray中填充ga数据字段
         *
         * @param array       ga数据字段
         * @param outputArray 输出数组
         */
        private void fillInGa(String[] array, String[] outputArray) throws IOException {
            outputArray[1] = array[1]; //device_type
            outputArray[9] = array[26]; //gender
            outputArray[10] = array[27]; //birth_year
//            outputArray[11] = ""; //profession
//            outputArray[12] = ""; //education level
//            outputArray[13] = ""; //income range
//            outputArray[14] = ""; //family status
//            outputArray[15] = ""; //hobby
            //outputArray[16:37] = array[4:25]
            System.arraycopy(array, 4, outputArray, 16, 22);
            outputArray[38] = array[28]; //revenue_week
            outputArray[39] = array[29]; //revenue_month
            outputArray[40] = array[30]; //average_session_per_app
            outputArray[41] = array[31]; //average_days_per_app
            outputArray[42] = array[32]; //installation list
            outputArray[44] = formatter.print(DateTime.now());
        }
    }

    public int run(String[] strings) throws Exception {
        Configuration conf = getConf();
        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");

        Job job = Job.getInstance(conf, "dmp device total");
        job.setJarByClass(DmpDeviceTotalMR.class);

        FileInputFormat.addInputPath(job, new Path(strings[0]));
        FileInputFormat.addInputPath(job, new Path(strings[1]));
        FileInputFormat.addInputPath(job, new Path(strings[2]));

        Path outputPath = new Path(strings[3]);
        FileSystem fileSystem = outputPath.getFileSystem(conf);
        if (fileSystem.exists(outputPath)) {
            fileSystem.delete(outputPath, true);
        }
        FileOutputFormat.setOutputPath(job, outputPath);
        FileOutputFormat.setCompressOutput(job, true);
        FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);

        job.setInputFormatClass(TextInputFormat.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setMapperClass(DmpDeviceMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(DmpDeviceReducer.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new Configuration(), new DmpDeviceTotalMR(), args));
/*
        String gaTotal = "00000B0A-46ED-4F9D-A101-7D8BF004CFF6\tidfa\tios\tIQ\t2016-09-07\t0\t\t1\t1\t0\t1\t0\t0\t0\t0\t0\t0\t0\t24\t36.0\t24\t36.0\t24\t36.0\t527708077401556\t\tfemale\t\t0.0\t0.0\t8.0\t1.0\t[{\"date\": \"20160907\", \"appName\": \"949498190\"}]\t20161006";
        String adnTotal = "00000B0A-46ED-4F9D-A101-7D8BF004CFF6\tidfa\t0\tandroid\t5.0\t0\tx6a\tvivo\t0\tCN\t0\t[{\"publisher_id\":\"6028\",\"app_version\":\"0\",\"app_id\":\"20660\",\"datetime\":\"2016-09-06 21:44:49\"}]\t2016-09-06 21:44:49";
        Iterable<Text> iterable = Lists.newArrayList(new Text(gaTotal), new Text(adnTotal));
        new DmpDeviceReducer().reduce(new Text("00000B0A-46ED-4F9D-A101-7D8BF004CFF6\tidfa"), iterable, null);*/
    }
}