<?xml version="1.0" encoding="UTF-8"?>

 <project xmlns="http://maven.apache.org/POM/4.0.0"

          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

     <modelVersion>4.0.0</modelVersion>

     <groupId>netease.bigdata.course</groupId>

     <artifactId>etl</artifactId>

     <version>1.0-SNAPSHOT</version>

     <dependencies>

         <dependency>

             <groupId>org.apache.hadoop</groupId>

             <artifactId>hadoop-client</artifactId>

             <version>2.7.6</version>

             <scope>provided</scope>

         </dependency>

         <dependency>

             <groupId>com.alibaba</groupId>

             <artifactId>fastjson</artifactId>

             <version>1.2.4</version>

         </dependency>

     </dependencies>

     <build>

         <sourceDirectory>src/main</sourceDirectory>

         <plugins>

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-assembly-plugin</artifactId>

                <configuration>

                    <descriptorRefs>

                        <descriptorRef>

                            jar-with-dependencies

                        </descriptorRef>

                    </descriptorRefs>

                </configuration>

                <executions>

                    <execution>

                        <id>make-assembly</id>

                        <phase>package</phase>

                        <goals>

                            <goal>single</goal>

                        </goals>

                    </execution>

                </executions>

            </plugin>

         </plugins>

     </build>

 </project>

package com.bigdata.etl.job;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
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.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.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;

public class ParseLogJob extends Configured implements Tool {
    //日志解析函数  (输入每一行的值)
    public static  Text parseLog(String row) throws ParseException {
        String[] logPart = StringUtils.split(row, "\u1111");
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        long timeLog = dateFormat.parse(logPart[0]).getTime();
        String activeName = logPart[1];
        JSONObject bizData=JSONObject.parseObject(logPart[2]);
        JSONObject logData = new JSONObject();

        logData.put("active_name",activeName);
        logData.put("time_log",timeLog);
        logData.putAll(bizData);
        return new Text(logData.toJSONString());
    }

    //输入key类型，输入value类型，输出。。(序列化类型)
    public static class LogMapper extends Mapper<LongWritable,Text,NullWritable,Text>{
        //输入key值 输入value值 map运行的上下文变量
        public void map(LongWritable key ,Text value ,Context context) throws IOException,InterruptedException{
            try {
                Text parseLog = parseLog(value.toString());
                context.write(null,parseLog);
            } catch (ParseException e) {
                e.printStackTrace();
            }

        }
    }

    public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration config = getConf();
        Job job= Job.getInstance(config);
        job.setJarByClass(ParseLogJob.class);
        job.setJobName("parseLog");
        job.setMapperClass(LogMapper.class);
        //设置reduce 为0
        job.setNumReduceTasks(0);
        //命令行第一个参数作为输入路径
        FileInputFormat.addInputPath(job,new Path(args[0]));
        //第二个参数  输出路径
        Path outPutPath = new Path(args[1]);
        FileOutputFormat.setOutputPath(job,outPutPath);
        //防止报错  删除输出路径
        FileSystem fs = FileSystem.get(config);
        if (fs.exists(outPutPath)){
            fs.delete(outPutPath,true);
        }
        if (!job.waitForCompletion(true)){
            throw new RuntimeException(job.getJobName()+"fail");
        }
        return 0;
    }
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new ParseLogJob(), args);
        System.exit(res);
    }
}