19.通过MAPREDUCE 把收集数据进行清洗

在eclipse软件里创建一个maven项目

jdk要换成本地安装的1.8版本的

加载pom.xml文件

<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>com.it19gong</groupId>

  <artifactId>clickLog</artifactId>

  <version>0.0.1-SNAPSHOT</version>

  <packaging>jar</packaging>

  <name>clickLog</name>

  <url>http://maven.apache.org</url>

  <properties>

    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

  </properties>

  <dependencies>

    <dependency>

      <groupId>junit</groupId>

      <artifactId>junit</artifactId>

      <version>3.8.1</version>

      <scope>test</scope>

    </dependency>

    <dependency>

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

            <artifactId>hadoop-common</artifactId>

            <version>2.6.0</version>

        </dependency>

        <dependency>

            <groupId>jdk.tools</groupId>

            <artifactId>jdk.tools</artifactId>

            <version>1.8</version>

            <scope>system</scope>

            <systemPath>E:/software/jdk1.8/lib/tools.jar</systemPath>

        </dependency>

        <dependency>

            <groupId>org.apache.hive</groupId>

            <artifactId>hive-jdbc</artifactId>

            <version>2.1.0</version>

        </dependency>

        <dependency>

    <groupId>mysql</groupId>

    <artifactId>mysql-connector-java</artifactId>

    <version>5.1.33</version>

      </dependency>

  </dependencies>

</project>

在加载依赖包的时候如果出现错误，在仓库里找不到1.8jdk.tools

在这个地方改成本地的jdk绝对路径，再重新加载一次maven的依赖包

我这里修改成

在项目下新建AccessLogPreProcessMapper类

package com.it19gong.clickLog;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

public class AccessLogPreProcessMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    Text text = new Text();

    @Override

protected void map(LongWritable key, Text value,Context context)

        throws IOException, InterruptedException {

       String itr[] = value.toString().split(" ");

       if (itr.length < 11)

        {

            return;

        }

        String ip = itr[0];

        String date = AnalysisNginxTool.nginxDateStmpToDate(itr[3]);

        String url = itr[6];

        String upFlow = itr[9];

        text.set(ip+","+date+","+url+","+upFlow);

        context.write(text, NullWritable.get());

}

}

创建AnalysisNginxTool类

package com.it19gong.clickLog;

import java.text.ParseException;

import java.text.SimpleDateFormat;

import java.util.Date;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

public class AnalysisNginxTool

{

    private static Logger logger = LoggerFactory.getLogger(AnalysisNginxTool.class);

    public static String nginxDateStmpToDate(String date)

    {

        String res = "";

        try

        {

            SimpleDateFormat df = new SimpleDateFormat("[dd/MM/yyyy:HH:mm:ss");

            String datetmp = date.split(" ")[0].toUpperCase();

            String mtmp = datetmp.split("/")[1];

            DateToNUM.initMap();

            datetmp = datetmp.replaceAll(mtmp, (String) DateToNUM.map.get(mtmp));

            System.out.println(datetmp);

            Date d = df.parse(datetmp);

            SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd");

            res = sdf.format(d);

        }

        catch (ParseException e)

        {

            logger.error("error:" + date, e);

        }

        return res;

    }

    public static long nginxDateStmpToDateTime(String date)

    {

        long l = 0;

        try

        {

            SimpleDateFormat df = new SimpleDateFormat("[dd/MM/yyyy:HH:mm:ss");

            String datetmp = date.split(" ")[0].toUpperCase();

            String mtmp = datetmp.split("/")[1];

            datetmp = datetmp.replaceAll(mtmp, (String) DateToNUM.map.get(mtmp));

            Date d = df.parse(datetmp);

            l = d.getTime();

        }

        catch (ParseException e)

        {

            logger.error("error:" + date, e);

        }

        return l;

    }

}

创建DateToNUM类

package com.it19gong.clickLog;

import java.util.HashMap;

public class DateToNUM

{

    public static HashMap map = new HashMap();

    public static void initMap()

    {

        map.put("JAN", "01");

        map.put("FEB", "02");

        map.put("MAR", "03");

        map.put("APR", "04");

        map.put("MAY", "05");

        map.put("JUN", "06");

        map.put("JUL", "07");

        map.put("AUG", "08");

        map.put("SEPT", "09");

        map.put("OCT", "10");

        map.put("NOV", "11");

        map.put("DEC", "12");

    }

}

新建AccessLogDriver类

package com.it19gong.clickLog;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class AccessLogDriver {

    public static void main(String[] args) throws Exception {

        DateToNUM.initMap();

        Configuration conf = new Configuration();

        if(args.length != 2){

            args = new String[2];

            args[0] =  "hdfs://node1/data/clickLog/20190620/";

            args[1]    =  "hdfs://node1/uvout/hive" ;

        }

        Job job = Job.getInstance(conf); // 设置一个用户定义的job名称

        job.setJarByClass(AccessLogDriver.class);

        job.setMapperClass(AccessLogPreProcessMapper.class); // 为job设置Mapper类

        // 为job设置Reducer类

        job.setNumReduceTasks(0);

        job.setMapOutputKeyClass(Text.class);// 为job的输出数据设置Key类

        job.setMapOutputValueClass(NullWritable.class);// 为job输出设置value类

        FileInputFormat.addInputPath(job, new Path(args[0])); // 为job设置输入路径

        FileOutputFormat.setOutputPath(job, new Path(args[1]));// 为job设置输出路径

        System.exit(job.waitForCompletion(true) ? 0 : 1); // 运行job

    }

}

把工程打包成Jar包

把jar包上传到集群

在集群上运行一下，先检查一下集群的启动进程

 hadoop jar mrclick.jar com.it19gong.clickLog.AccessLogDriver

可以看到输出目录

查看清洗后的数据

巴特西

19.通过MAPREDUCE 把收集数据进行清洗

最新文章

热门文章