十五、Hadoop中数据清洗（ETL）的简单应用

青旅半醒 2021-09-01 09:26 456阅读 0赞

数据清洗，是每个业务中不可或缺的部分，在运行核心业务的MapReduce程序之前，往后都会对数据进行清洗。数据清洗的过程往往只需要运行Mapper程序，而不需要运行Reducer程序，本文主要介绍一下数据清洗的简单应用。关注专栏[《破茧成蝶——大数据篇》][Link 1]查看相关系列的文章~

--------------------

**目录**

一、开始的话

二、需求与数据

三、定义Bean类

四、编写Mapper类

五、编写Driver驱动类

六、测试

--------------------

# 一、开始的话  #

因为是简单应用，本文主要使用一个实例来介绍数据的ETL清洗。值得一提的是，在企业中，几乎都会有专门的数据清洗工具，像本文介绍的方法或许用的并不是很多。废话不多说了，来看个简单的应用示例吧。

# 二、需求与数据 #

还是依然用咱们之前的Nginx的日志数据吧。数据形式如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70][]

字段依次是：时间、版本、客户端ip、访问路径、状态、域名、服务端ip、size、响应时间。现在需要对数据进行清洗，将访问请求不为200的过滤掉，并且将每行日志字段数小于5的过滤掉，字段数小于5，我们认为是对我们没有用的数据。

# 三、定义Bean类 #

package com.xzw.hadoop.mapreduce.etl;
    
    /**
     * @author: xzw
     * @create_date: 2020/8/18 13:38
     * @desc:
     * @modifier:
     * @modified_date:
     * @desc:
     */
    public class ETLLogBean {
        private String date;
        private String version;
        private String clientIP;
        private String url;
        private String status;
        private String domainName;
        private String serverIP;
        private String size;
        private String responseDate;
    
        private boolean valid = true;//判断数据是否合法
    
        @Override
        public String toString() {
    
            StringBuilder sb = new StringBuilder();
    
            sb.append(this.valid);
            sb.append("\001").append(this.date);
            sb.append("\001").append(this.version);
            sb.append("\001").append(this.clientIP);
            sb.append("\001").append(this.url);
            sb.append("\001").append(this.status);
            sb.append("\001").append(this.domainName);
            sb.append("\001").append(this.serverIP);
            sb.append("\001").append(this.size);
            sb.append("\001").append(this.responseDate);
    
            return sb.toString();
        }
    
        public String getDate() {
            return date;
        }
    
        public void setDate(String date) {
            this.date = date;
        }
    
        public String getVersion() {
            return version;
        }
    
        public void setVersion(String version) {
            this.version = version;
        }
    
        public String getClientIP() {
            return clientIP;
        }
    
        public void setClientIP(String clientIP) {
            this.clientIP = clientIP;
        }
    
        public String getUrl() {
            return url;
        }
    
        public void setUrl(String url) {
            this.url = url;
        }
    
        public String getStatus() {
            return status;
        }
    
        public void setStatus(String status) {
            this.status = status;
        }
    
        public String getDomainName() {
            return domainName;
        }
    
        public void setDomainName(String domainName) {
            this.domainName = domainName;
        }
    
        public String getServerIP() {
            return serverIP;
        }
    
        public void setServerIP(String serverIP) {
            this.serverIP = serverIP;
        }
    
        public String getSize() {
            return size;
        }
    
        public void setSize(String size) {
            this.size = size;
        }
    
        public String getResponseDate() {
            return responseDate;
        }
    
        public void setResponseDate(String responseDate) {
            this.responseDate = responseDate;
        }
    
        public boolean isValid() {
            return valid;
        }
    
        public void setValid(boolean valid) {
            this.valid = valid;
        }
    }

这里说明一下，在上面的代码的toString方法中用到了如下的一个分隔符：

sb.append("\001").append(this.date);

这个\\001的分隔符是Hive中的默认分隔符，Hive会在后面的课程中进行讲解，这里先知道这么回事就行。

# 四、编写Mapper类 #

package com.xzw.hadoop.mapreduce.etl;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    /**
     * @author: xzw
     * @create_date: 2020/8/18 13:44
     * @desc:
     * @modifier:
     * @modified_date:
     * @desc:
     */
    public class ETLLogMapper02 extends Mapper<LongWritable, Text, Text, NullWritable> {
        Text k = new Text();
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
    
            //解析日志是否合法
            ETLLogBean bean = parseLog(line);
    
            if (!bean.isValid())
                return;
    
            k.set(bean.toString());
    
            //输出
            context.write(k, NullWritable.get());
        }
    
        /**
         * 解析日志
         * @param line
         * @return
         */
        private ETLLogBean parseLog(String line) {
            ETLLogBean logBean = new ETLLogBean();
    
            String[] fields = line.split("\t");
    
            if (fields.length > 5) {
                logBean.setDate(fields[0]);
                logBean.setVersion(fields[1]);
                logBean.setClientIP(fields[2]);
                logBean.setUrl(fields[3]);
                logBean.setStatus(fields[4]);
                logBean.setDomainName(fields[5]);
                logBean.setServerIP(fields[6]);
                logBean.setSize(fields[7]);
                logBean.setResponseDate(fields[8]);
    
                if (Integer.parseInt(logBean.getStatus()) >= 400) {
                    logBean.setValid(false);
                }
            } else {
                logBean.setValid(false);
            }
            return logBean;
        }
    }

# 五、编写Driver驱动类 #

package com.xzw.hadoop.mapreduce.etl;
    
    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;
    
    import java.io.IOException;
    
    /**
     * @author: xzw
     * @create_date: 2020/8/17 9:31
     * @desc:
     * @modifier:
     * @modified_date:
     * @desc:
     */
    public class ETLLogDriver {
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            args = new String[]{"e:/input/nginx_log", "e:/output"};
    
            Job job = Job.getInstance(new Configuration());
    
            job.setJarByClass(ETLLogDriver.class);
    
            job.setMapperClass(ETLLogMapper02.class);
    
            job.setNumReduceTasks(0);
    
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);
    
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
            boolean b = job.waitForCompletion(true);
    
            System.exit(b ? 0 : 1);
        }
    }

# 六、测试 #

运行，看一下效果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1][]

这就是咱们清洗过的数据。再多说一句吧，在后台运行的日志中，有这样的一段

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2][]

这是Hadoop内置的计数器，我们可以通过这些数据，来描述多项指标。例如：可以监控已经处理的输入数据量和输出数据量等等。

本文比较简单，你们在这个过程中遇到了什么问题，欢迎留言，让我看看你们遇到了什么问题~

[Link 1]: https://blog.csdn.net/gdkyxy2013/category_10006538.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70]: /images/20210726/6767b97b2d5048daa6c57c7c451cb792.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1]: /images/20210726/087c751e0e954301a2271067748b48b7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2]: /images/20210726/0983fdad4862443f81474cde8cc26d88.png