实现自定义的InputFormat

我们可以为MapReduce计算实现并指定自定义的InputFormat实现，来获得对输入数据更多的控制，以及支持专有的或特定应用程序的输入数据文件格式作为Hadoop MapReduce计算的输入。

自定义的InputFormat实现应该继承org.apache.hadoop.mapreduce.InputFormat<K,V> 抽象类，并重写createRecordReader()和getSplits()方法。

需求描述

在这一节，我们为HTTP日志文件实现一个自定义的InputFormat和RecordReader。这个InputFormat将生成LongWritable类型的key和LogWritable类型的value。

一、创建Java Maven项目

Maven依赖：

<?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>HadoopDemo</groupId>
    <artifactId>com.xueai8</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <!--hadoop依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.3.1</version>
        </dependency>
        <!--hdfs文件系统依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.3.1</version>
        </dependency>
        <!--MapReduce相关的依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
            <version>3.3.1</version>
        </dependency>
        <!--junit依赖-->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!--编译器插件用于编译拓扑-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <!--指定maven编译的jdk版本和字符集,如果不指定,maven3默认用jdk 1.5 maven2默认用jdk1.3-->
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source> <!-- 源代码使用的JDK版本 -->
                    <target>1.8</target> <!-- 需要生成的目标class文件的编译版本 -->
                    <encoding>UTF-8</encoding><!-- 字符集编码 -->
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

LogWritable.java：

这是自定义value数据类型，需要实现Writable接口

package com.xueai8.custominput;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

/**
 *
 * 自定义value数据类型，需要实现Writable接口
 */
public class LogWritable implements Writable {
    private Text userIP;		// 客户端的IP地址
    private Text timestamp;		// 客户访问时间
    private Text url;			// 客户访问的url
    private IntWritable status;         // 状态码
    private IntWritable responseSize;	// 服务端响应数据的大小

    public LogWritable() {
        this.userIP = new Text();
        this.timestamp = new Text();
        this.url = new Text();
        this.status = new IntWritable();
        this.responseSize = new IntWritable();
    }

    public void set(String userIP, String timestamp, String url, int status, int responseSize) {
        this.userIP.set(userIP);
        this.timestamp.set(timestamp);
        this.url.set(url);
        this.status.set(status);
        this.responseSize.set(responseSize);
    }

    public Text getUserIP() {
        return userIP;
    }

    public void setUserIP(Text userIP) {
        this.userIP = userIP;
    }

    public Text getTimestamp() {
        return timestamp;
    }

    public void setTimestamp(Text timestamp) {
        this.timestamp = timestamp;
    }

    public Text getUrl() {
        return url;
    }

    public void setUrl(Text url) {
        this.url = url;
    }

    public IntWritable getStatus() {
        return status;
    }

    public void setStatus(IntWritable status) {
        this.status = status;
    }

    public IntWritable getResponseSize() {
        return responseSize;
    }

    public void setResponseSize(IntWritable responseSize) {
        this.responseSize = responseSize;
    }

    // 序列化方法
    @Override
    public void write(DataOutput out) throws IOException {
        userIP.write(out);
        timestamp.write(out);
        url.write(out);
        status.write(out);
        responseSize.write(out);
    }

    // 反序列化方法
    @Override
    public void readFields(DataInput in) throws IOException {
        userIP.readFields(in);
        timestamp.readFields(in);
        url.readFields(in);
        status.readFields(in);
        responseSize.readFields(in);
    }

}

LogFileRecordReader.java：

这是自定义的RecordReader实现的实例。

package com.xueai8.custominput;

import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;

/**
 *
 * 自定义RecordReader，读取输入的日志文件记录，按正则表达式模式提取各字段
 * 并封装到自定义的LogWritable value类型中
 */
public class LogFileRecordReader extends RecordReader<LongWritable, LogWritable> {

    private LineRecordReader lineReader;    // 默认RecordReader
    private LogWritable value;              // 读取转换为key-value中的value

    @Override
    public void initialize(InputSplit inputSplit, TaskAttemptContext attempt)
            throws IOException, InterruptedException {
        lineReader = new LineRecordReader();
        lineReader.initialize(inputSplit, attempt);
    }

    // 定制RR如何构造value结构(这里将value构造为LogWritable类型)
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if (!lineReader.nextKeyValue()) {
            return false;
        }

        // 正则表达式
        String logEntryPattern = "^(\\S+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3}) (\\d+)";

        // 使用正则表达式从日志项中提取字段
        Pattern p = Pattern.compile(logEntryPattern);
        Matcher matcher = p.matcher(lineReader.getCurrentValue().toString());
        if (!matcher.matches()) {
            System.out.println("Bad Record:" + lineReader.getCurrentValue());
            return nextKeyValue();
        }

        String userIP = matcher.group(1);
        String timestamp = matcher.group(4);
        String request = matcher.group(5);
        int status = Integer.parseInt(matcher.group(6));
        int bytes = Integer.parseInt(matcher.group(7));

        value = new LogWritable();
        value.set(userIP, timestamp, request, bytes, status);
        return true;
    }

    @Override
    public LongWritable getCurrentKey() throws IOException, InterruptedException {
        return lineReader.getCurrentKey();
    }

    @Override
    public LogWritable getCurrentValue() throws IOException, InterruptedException {
        return value;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return lineReader.getProgress();
    }

    @Override
    public void close() throws IOException {
        lineReader.close();
    }
}

LogFileInputFormat.java：

实现自定义的LogFileInputFormat类，它继承自FileInputFormat类，用于操作HDFS文件中的数据。

package com.xueai8.custominput;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

/**
 *
 * 实现自定义的LogFileInputFormat类，它继承自FileInputFormat类，用于操作HDFS文件中的数据
 */
public class LogFileInputFormat extends FileInputFormat<LongWritable, LogWritable> {

    @Override
    public RecordReader<LongWritable, LogWritable> createRecordReader(InputSplit arg0, TaskAttemptContext arg1)
            throws IOException, InterruptedException {
        return new LogFileRecordReader();
    }

}

LogMapper.java：

Mapper实现。

package com.xueai8.custominput;

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 *
 * Mapper实现。确保该计算的Mapper使用LongWritable作为input key类型，LogWritable作为input value类型
 */
public class LogMapper extends Mapper<Object, LogWritable, Text, LogWritable > {

    @Override
    public void map(Object key, LogWritable value, Context context)
            throws IOException, InterruptedException {
        // 写出，以ip为key，LogWritable为value
        context.write(value.getUserIP(),value);
    }
}

LogReducer.java：

Reducer实现。

package com.xueai8.custominput;

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 *
 * Reducer实现。输入的值类型与Mapper的输出值类型保持一致
 */
public class LogReducer extends Reducer<Text,LogWritable,Text,IntWritable> {

    private IntWritable result = new IntWritable();

    @Override
    public void reduce(Text key, Iterable<LogWritable> values, Context context)
            throws IOException, InterruptedException {
        int sum = 0;
        for (LogWritable val : values) {
            sum += val.getResponseSize().get();
        }
        result.set(sum);
        context.write(key, result);
    }
}

LogDriver.java：

驱动程序，接收任何文本输入文件，使用自定义的LogFileInputFormat作为MapReduce计算的InputFormat。

package com.xueai8.custominput;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;

/**
 *
 * 使用Job对象指定LogFileInputFormat作为MapReduce计算的InputFormat
 */
public class LogDriver {

    public static void main(String[] args) throws Exception{
        if (args.length < 2) {
            // System.err.println("语法:  <input_path> <output_path> <num_reduce_tasks>");
            System.err.println("语法:  <input_path> <output_path>");
            System.exit(-1);
        }

        // set job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "log-analysis");
        job.setJarByClass(LogDriver.class);

        // set mapper
        job.setMapperClass(LogMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LogWritable.class);

        // set reducer
        job.setReducerClass(LogReducer.class);
        // int numReduce = Integer.parseInt(args[2]);
        // job.setNumReduceTasks(numReduce);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setInputFormatClass(LogFileInputFormat.class);   /* 指定使用自定义的InputFormat */

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.out.println(job.waitForCompletion(true) ? 0 : 1);
    }
}

代码说明

LogFileInputFormat继承自FileInputFormat, 而FileInputFormat为基于HDFS文件的InputFormat提供了一个通用的Splitting机制。

我们在LogFileInputFormat中重写createRecordReader()方法，以提供一个我们自定义的RecordReader实现的实例-LogFileRecordReader。

可选地，我们还可以重写FileInputFormat类的isSplitable()方法以控制该input文件是否被切分为逻辑分区或作为整个文件使用。

public RecordReader<LongWritable, LogWritable> createRecordReader(InputSplit arg0, TaskAttemptContext arg1) throws …… {
    return new LogFileRecordReader();
}

这个 LogFileRecordReader类继承自org.apache.hadoop.mapreduce.RecordReader<K,V>抽象类，内部使用LineRecordReader来执行输入数据的基本解析。

lineReader = new LineRecordReader();
lineReader.initialize(inputSplit, attempt);

我们在nextKeyValue()方法中执行输入数据的日志项的自定义解析。使用正则表达来抽取HTTP服务器日志项的字段，并使用这些字段填充一个LogWritable类的实例。

public boolean nextKeyValue() throws IOException, ..{
    if (!lineReader.nextKeyValue())
        return false;

    String line = lineReader.getCurrentValue().toString();
    ……………		// 使用正则表达式从line中提取字段
    value = new LogWritable(userIP, timestamp, request, status, bytes);
    return true;
}

我们可以通过重写InputFormat类的getSplits()方法来执行自定义输入数据的切分。这个getSplits()方法应该返回一个InputSplit对象的列表。一个InputSplit对象代表输入数据的一个逻辑分区，并且会被赋给一个单独的Map任务。

InputSplit类继承自InputSplit抽象类，需要重写getLocations()和getLength()方法。其中getLength()方法应该提供该split的长度，getLocations()方法应该提供这个split所代表的数据的物理存储节点列表。Hadoop使用一个数据本地化节点列表用于Map任务调度。

我们在前面的示例中所使用的FileInputFormat类使用org.apache.hadoop.mapreduce.lib.input.FileSplit作为InputSplit实现。

我们也可以为非HDFS数据编写InputFormat实现。例如，支持从一个SQL表读取输入数据的DBInputFormat，org.apache.hadoop.mapreduce.lib.db.DBInputFormat就是InputFormat的例子。

二、配置log4j

在src/main/resources目录下新增log4j的配置文件log4j.properties，内容如下：

log4j.rootLogger = info,stdout

### 输出信息到控制抬 ###
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n

三、项目打包

打开IDEA下方的终端窗口terminal，执行"mvn clean package"打包命令，如下图所示：

如果一切正常，会提示打jar包成功。如下图所示：

这时查看项目结构，会看到多了一个target目录，打好的jar包就位于此目录下。如下图所示：

四、项目部署

请按以下步骤执行。

1、启动HDFS集群和YARN集群。在Linux终端窗口中，执行如下的脚本：

$ start-dfs.sh
$ start-yarn.sh

查看进程是否启动，集群运行是否正常。在Linux终端窗口中，执行如下的命令：

$ jps

这时应该能看到有如下5个进程正在运行，说明集群运行正常：

    5542 NodeManager
    5191 SecondaryNameNode
    4857 NameNode
    5418 ResourceManager
    4975 DataNode

2、将日志数据文件log_sample.txt上传到HDFS的/data/mr/目录下。

$ hdfs dfs -mkdir -p /data/mr
$ hdfs dfs -put log_sample.txt /data/mr/
$ hdfs dfs -ls /data/mr/

3、提交作业到Hadoop集群上运行。（如果jar包在Windows下，请先拷贝到Linux中。）

在终端窗口中，执行如下的作业提交命令：

$ hadoop jar HadoopDemo-1.0-SNAPSHOT.jar com.xueai8.custominput.LogDriver /data/mr /data/mr-output

4、查看输出结果。

在终端窗口中，执行如下的HDFS命令，查看输出结果：

$ hdfs dfs -ls /data/mr-output 
$ hdfs dfs -cat /data/mr-output/part-r-00000

可以看到最后的统计结果如下：

129.94.144.152	504
199.120.110.21	600
199.72.81.55	1400
205.189.154.54	1502
205.212.115.106	400
alyssa.prodigy.com	200
burger.letters.com	808
d104.aa.net	800
dave.dev1.ihub.com	800
dd14-012.compuserve.com	200
dial22.lloyd.com	200
gater3.sematech.org	400
gater4.sematech.org	400
gayle-gaston.tenet.edu	200
ix-or10-06.ix.netcom.com	400
ix-orl2-01.ix.netcom.com	600
link097.txdirect.net	1800
net-1-141.eden.com	200
netport-27.iu.net	200
onyx.southwind.net	704
piweba3y.prodigy.com	400
pm13.j51.com	200
port26.annex2.nwlink.com	1800
ppp-mia-30.shadow.net	1200
ppp-nyc-3-1.ios.com	400
ppptky391.asahi-net.or.jp	400
remote27.compusmart.ab.ca	902
scheyer.clark.net	200
slip1.yab.com	400
smyth-pc.moorecap.com	600
unicomp6.unicomp.net	1000
waters-gw.starway.net.au	200
www-a1.proxy.aol.com	200
www-b4.proxy.aol.com	200

小白学苑

让大数据学习更简单