Hadoop实战 – Apache访问日志

本示例通过mapreduce计算任务实现对apache访问日志的解析存储和分析，将分析记录存储于HBase数据库中，该示例只是一个日志处理环节，后续处理可进一步扩充。

日志说明

在apache目录配置文件（/etc/httpd/conf/httpd.conf）中有如下配置：

httpd.conf

# The following directives define some format nicknames for use with
# a CustomLog directive (see below).
#
LogFormat "%h %l %u %t \"%r\" %>s %b \"%{Referer}i\" \"%{User-Agent}i\"" combined
LogFormat "%h %l %u %t \"%r\" %>s %b" common
LogFormat "%{Referer}i -> %U" referer
LogFormat "%{User-agent}i" agent
# For a single logfile with access, agent, and referer information
# (Combined Logfile Format), use the following directive:
#
CustomLog logs/access_log combined

该配置启用了名为combined的日志格式

该格式为%h %l %u %t \"%r\" %>s %b \"%{Referer}i\" \"%{User-Agent}i\"

以下是官方对这些格式字段的描述：

%%	百分号(Apache2.0.44或更高的版本)
%a	远端IP地址
%A	本机IP地址
%B	除HTTP头以外传送的字节数
%b	以CLF格式显示的除HTTP头以外传送的字节数，也就是当没有字节传送时显示’-‘而不是0。
%{Foobar}C	在请求中传送给服务端的cookieFoobar的内容。
%D	服务器处理本请求所用时间，以微为单位。
%{FOOBAR}e	环境变量FOOBAR的值
%f	文件名
%h	远端主机
%H	请求使用的协议
%{Foobar}i	发送到服务器的请求头Foobar:的内容。
%l	远端登录名(由identd而来，如果支持的话)，除非IdentityCheck设为”On”，否则将得到一个”-“。
%m	请求的方法
%{Foobar}n	来自另一个模块的注解Foobar的内容。
%{Foobar}o	应答头Foobar:的内容。
%p	服务器服务于该请求的标准端口。
%P	为本请求提供服务的子进程的PID。
%{format}P	服务于该请求的PID或TID(线程ID)，format的取值范围为：pid和tid(2.0.46及以后版本)以及hextid(需要APR1.2.0及以上版本)
%q	查询字符串(若存在则由一个”?”引导，否则返回空串)
%r	请求的第一行
%s	状态。对于内部重定向的请求，这个状态指的是原始请求的状态，—%>s则指的是最后请求的状态。
%t	时间，用普通日志时间格式(标准英语格式)
%{format}t	时间，用strftime(3)指定的格式表示的时间。(默认情况下按本地化格式)
%T	处理完请求所花时间，以秒为单位。
%u	远程用户名(根据验证信息而来；如果返回status(%s)为401，可能是假的)
%U	请求的URL路径，不包含查询字符串。
%v	对该请求提供服务的标准ServerName。
%V	根据UseCanonicalName指令设定的服务器名称。
%X	请求完成时的连接状态：X= 连接在应答完成前中断。 += 应答传送完后继续保持连接。 -= 应答传送完后关闭连接。 (在1.3以后的版本中，这个指令是%c，但这样就和过去的SSL语法：%{var}c冲突了)
%I	接收的字节数，包括请求头的数据，并且不能为零。要使用这个指令你必须启用mod_logio模块。
%O	发送的字节数，包括请求头的数据，并且不能为零。要使用这个指令你必须启用mod_logio模块。

从上表可得该日志的格式表示为：

• 客户端的IP地址。
• 由客户端identd进程判断的RFC1413身份(identity),输出中的符号”-“表示此处的信息无效。
• HTTP认证系统得到的访问该网页的客户标识(userid),如果网页没有设置密码保护，则此项将是”-“。
• 服务器完成请求处理时的时间。
• 客户的动作\请求的资源\使用的协议。
• 服务器返回给客户端的状态码。
• 返回给客户端的不包括响应头的字节数.如果没有信息返回，则此项应该是”-“。
• “Referer”请求头。
• “User-Agent”请求头。

例如：

103.226.133.67 - - [22/Sep/2016:21:00:39 +0800] "GET /hbase-distribute/ HTTP/1.1" 200 35477 "-" "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:48.0) Gecko/20100101 Firefox/48.0"

任务分析

使用hadoop MapReduce任务解析日志并存入Hbase。

功能需求

• 日志解析
• 存储
• 分析

业务步骤

• 读取日子记录
• 解析日志记录
• 存储已解析内容
• 读取已解析内容
• 分析数据
• 存储结果

如图，分两个mapreduce来完成，第一个只有map用于解析，第二个使用map加载，reduce计数。后续的分析也依赖前面的解析操作。

HTable表设计

由于只是存储网站每小时（RowKey）访问流量信息，故只需一个表示数量的列簇，同时，为了网站访问分析（比如热点分析），增加一个列簇表示最大流量的标识（如连接地址、访问IP等）。

此处是预先设计的HTable结构的层次示意，Table结构如下：

access-info
  |--datetime(key)            #日期时间为RowKey
  |    |--counter             #流量计数列簇
  |    |    |--page           #页面访问量列
  |    |    |--user           #用户访问量列
  |    |    |--post           #POST请求量列
  |    |    |--get            #GET请求量列
  |    |    |--put            #PUT请求量列
  |    |    |--delete         #DELETE请求量列
  |    |    |--error          #错误解析（一般认为脚本攻击）列
  |    |    |--connect
  |    |    |--other
  |    |    ...
  |    |--max                 #最多访问列簇
  |    |    |--page           #最多访问页列
  |    |    |--ip             #最多IP访问列

使用命令以创建该表（或在程序中运行创建）

create 'access-info',{NAME=>'counter',VERSIONS=>1},{NAME=>'max',VERSIONS=>10}

编程实现

初始部分

日志记录在整个分析中，起到POJO的作用，因此先封装一个日志记录对象

LogRecord.java

public class LogRecord {
    private static final SimpleDateFormat format = new SimpleDateFormat("d/MMM/yy:HH:mm:ss Z", Locale.ENGLISH);
    private String ip;
    private String loginName;
    private String userName;
    private Date dateTime;
    private String method;
    private String url;
    private String protocol;
    private int stateCode;
    private long responseLength;
    private String referFrom;
    private String userAgent;
    private String unhandled;
    // 以下省去get/set方法
    // ...
}

代码定义了日志记录的每一个元素属性，为了便于日期格式的转换，静态定义了一个私有SimpleDateFormat，同时额外增加了一个unhandled属性用于存储异常的格式解析。

日志的解析首先应该想到字符串的截取操作，匹配模式，对，就是正则表达式

LogAnalyser.java

public class LogAnalyser {
    /*
    正则的一些说明
    ^ :匹配每一行的开头。
    ^([0-9.]+)\s :匹配IP地址，\s匹配不可见字符，例如空格、制表符等。
    ([\w."-]+)\s :匹配identity，由下划线在内的任何单词字符（包括数字）或点引号杠组成。
    ([\w."-]+)\s :匹配userid。
    \[([^\[\]]+)\]\s :匹配时间，匹配外围的中括号且内部不存在中括号。
    "(.+)"\s :匹配请求信息，匹配外围双引号，匹配内部任意字符。
    (\d{3})\s :匹配状态码，三个长度的数字。
    (\d+|-)\s :匹配响应字节数或-。
    "((?:[^"]|\"-)*)"\s :匹配"Referer"请求头，双引号中可能出现转义的双引号\"。
    "((?:[^"]|\"-)*)" :匹配"User-Agent"请求头。
    $ :匹配行尾。
     */
    private static final String regx = "^([0-9.]+)\\s([\\w.\"-]+)\\s([\\w.\"-]+)\\s\\[([^\\[\\]]+)\\]\\s\"(.+)\"\\s(\\d{3})\\s(\\d+|-)\\s\"((?:[^\"]|\\\"-)*)\"\\s\"((?:[^\"]|\\\"-)*)\"$";
    private static final Pattern pattern = Pattern.compile(regx);
    public static LogRecord read(String source){
        Matcher matcher = pattern.matcher(source);
        if(!matcher.matches()|| matcher.groupCount()!=9){
            throw new IllegalArgumentException("error input format, source is : "+source);
        }else {
            LogRecord record = new LogRecord();
            record.setIp(matcher.group(1));
            record.setLoginName(matcher.group(2).replaceAll("-",""));
            record.setUserName(matcher.group(3).replaceAll("-",""));
            record.setDateTime(matcher.group(4));
            String[] tmp = matcher.group(5).split(" ");
            if(tmp.length!=3){
                record.setUrl(matcher.group(5));
                //throw new IllegalArgumentException("the %r (RequestHeader) format error, source is : "+source);
            }else {
                record.setMethod(tmp[0]);
                record.setUrl(tmp[1]);
                record.setProtocol(tmp[2]);
            }
            record.setStateCode(Integer.parseInt(matcher.group(6).replace('-','0')));
            record.setResponseLength(Long.parseLong(matcher.group(7).replace('-','0')));
            record.setReferFrom(matcher.group(8));
            record.setUserAgent(matcher.group(9));
            return record;
        }
    }
}

如上代码，通过正则表达式匹配，获取各匹配组，按顺序赋值到日志记录对象。但注意，此处有坑，如果正则表达式效率不高，则会在一些匹配组长度上出现堆栈溢出，了解正则匹配原理后对正则表达式优化才能解决该问题。但实际问题实际对待，由于日志记录的解析操作仅仅是字符串的拆分，如果没有很优秀的正则技巧，考虑字符串截取处理会更高效，针对该日志类型，记录字符串起始段有固定格式（主机IP、标识、用户、固定格式的时间），而中间段（请求信息）具有不确定性，因为在大量日志记录分析中看到，该段内容一般为较长且包含大量不规则字符（包括各种符号）的连接，也有可能为一堆脚本（站点扫描、检测、攻击等），但比起尾段（状态码、字节数、Refer、User-Agent）更加具有不可预见性，所以在字符串截取时应该优先处理格式可知的字段，从两端向中间截取，最后剩下不可预知的中间段，对其单独处理即可。

解析后的数据如何存储？

需要按时段分析，每小时的网站流量，那么利用mapreduce的shuffle按时间（精确到小时）排序存储即可，所以键确定就是时间字段，值则是解析结果。对于结果的存储，由于结果是一个PO对象，可以考虑序列化存储，使用BytesWritable？可以考虑，但我更倾向于封装适合自己的，于是新建一个LogRecordWritable，实现Writable接口即可，序列化使用protostuff，比protobuf使用简单，尤其是不用自己写schema。

LogRecordWritable.java

public class LogRecordWritable implements Writable {
    private static final RuntimeSchema<LogRecord> schema = RuntimeSchema.createFrom(LogRecord.class);
    private LogRecord record;
    private BytesWritable data = new BytesWritable();
    public LogRecord getRecord() {
        return record;
    }
    public void read(byte[] data){
        record = schema.newMessage();
        ProtostuffIOUtil.mergeFrom(data,record,schema);
    }
    public void setRecord(LogRecord record) {
        this.record = record;
    }
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        byte[] data = ProtostuffIOUtil.toByteArray(record,schema, LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE));
        this.data.set(data,0,data.length);
        this.data.write(dataOutput);
    }
    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.data.readFields(dataInput);
        this.data.setCapacity(this.data.getLength());
        byte[] data = this.data.getBytes();
        record = schema.newMessage();
        ProtostuffIOUtil.mergeFrom(data,record,schema);
    }
}

巨坑预警，使用序列化的数据写入DataOutput和读取DataInput会出现数据不一致的情况，读取数据明显会比写入的多，需要先setCapacity再读取字节反序列化，原因参考接口的实现类（DataInputStram）源码，这里就不废话了。

解析存储

第一个mapreduce任务，只有mapper，将一条条的字符串记录解析为日期时间和LogRecordWritable的键值对，存储为序列化文件（便于压缩处理和传输）

Extracter.java

public static class ExtractMapper extends Mapper<Object, Text, Text, LogRecordWritable> {
        final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd-HH");
        LogRecordWritable logRecordWritable = new LogRecordWritable();
        Text date = new Text();
        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String source = value.toString();
            if (source != null && source.length() > 0) {
                try {
                    LogRecord record = LogAnalyser.read(source);
                    date.set(format.format(record.getDateTime()));
                    logRecordWritable.setRecord(record);
                } catch (Exception e) {
                    LogRecord record = new LogRecord();
                    record.setUnhandled(source);
                    logRecordWritable.setRecord(record);
                    date.clear();
                    date.set("error");
                }
                context.write(date, logRecordWritable);
            }
        }
}

使用精确到小时的时间字符串格式，这里要考虑其他情况，比如脚本攻击产生的访问记录就有点不规则，归类到error列。

分析存储

第二个mapreduce任务，mapper加载上一个任务解析的结果，reducer统计流量数并存储到HBase数据库

Loader.java

public static class LoadMapper extends Mapper<Text, LogRecordWritable, Text, LogRecordWritable> {
    @Override
    protected void map(Text key, LogRecordWritable value, Context context) throws IOException, InterruptedException {
        context.write(key,value);
    }
}
public static class CounterReducer extends TableReducer<Text, LogRecordWritable, ImmutableBytesWritable>{
    @Override
    protected void reduce(Text key, Iterable<LogRecordWritable> values, Context context) throws IOException, InterruptedException {
        Map<String,Integer> count = new HashMap<String, Integer>();
        Map<String,Integer> users = new HashMap<String, Integer>();
        Map<String,Integer> pages = new HashMap<String, Integer>();
        int pv=0;
        for(LogRecordWritable writable:values){
            String mk=null;
            LogRecord record = writable.getRecord();
            if(record.getUnhandled()!=null){
                mk="error";
            }else if(record.getUrl().startsWith("/")){
                users.put(record.getIp(),users.get(record.getIp())!=null?(users.get(record.getIp())+1):1);
                pages.put(record.getUrl(),pages.get(record.getUrl())!=null?(pages.get(record.getUrl())+1):1);
                mk=record.getMethod()!=null?record.getMethod().toLowerCase():null;
                pv++;
            }else {
                mk="other";
            }
            count.put(mk,count.get(mk)!=null?(count.get(mk)+1):1);
        }
        Put put = new Put(key.getBytes());
        for(String column : count.keySet()){
            if(column!=null){
                put.addColumn(Bytes.toBytes("counter"),Bytes.toBytes(column),Bytes.toBytes(String.valueOf(count.get(column))));
            }
        }
        int uv = users.size();
        put.addColumn(Bytes.toBytes("counter"),Bytes.toBytes("user"),Bytes.toBytes(String.valueOf(uv)));
        put.addColumn(Bytes.toBytes("counter"),Bytes.toBytes("page"),Bytes.toBytes(String.valueOf(pv)));
        put.addColumn(Bytes.toBytes("max"),Bytes.toBytes("page"),Bytes.toBytes(max(pages)));
        put.addColumn(Bytes.toBytes("max"),Bytes.toBytes("ip"),Bytes.toBytes(max(users)));
        context.write(new ImmutableBytesWritable(key.getBytes()),put);
    }
    private String max(Map<String,Integer> map){
        int val = 0;
        String max = "-";
        for(String key:map.keySet()){
             if(map.get(key)>val){
                val = map.get(key);
                max = key;
             }
        }
        return max;
    }
}

此处有坑，将结果统一转换为字符串再转Bytes存储，否则就需要区别对待，直接全Bytes.toString()是不行的。

主函数实现

Parser.java

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        properties = new Properties();
        try {
            properties.load(Parser.class.getClassLoader().getResourceAsStream("conf.properties"));
        } catch (IOException e) {
            logger.error(e.getMessage(),e);
            System.exit(1);
        }
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS",properties.getProperty("fs.defaultFS"));
        conf.set("hbase.zookeeper.quorum",  properties.getProperty("hbase.zookeeper.quorum"));
        if(args.length>1) {
            Job analysis = Job.getInstance(conf);
            analysis.setMapperClass(Extracter.ExtractMapper.class);
            analysis.setOutputKeyClass(Text.class);
            analysis.setOutputValueClass(LogRecordWritable.class);
            analysis.setJarByClass(Parser.class);
            analysis.setOutputFormatClass(SequenceFileOutputFormat.class);
            analysis.setNumReduceTasks(0);
            FileInputFormat.addInputPath(analysis, new Path(args[0]));
            FileOutputFormat.setOutputPath(analysis, new Path(args[1]));
            Job counter = Job.getInstance(conf);
            counter.setMapperClass(Extracter.LoadMapper.class);
            counter.setMapOutputKeyClass(Text.class);
            counter.setMapOutputValueClass(LogRecordWritable.class);
            counter.setJarByClass(Parser.class);
            counter.setInputFormatClass(SequenceFileInputFormat.class);
            counter.setNumReduceTasks(1);
            TableMapReduceUtil.initTableReducerJob("access-info", Loader.CounterReducer.class, counter);
            FileInputFormat.addInputPath(counter, new Path(args[1]));
            JobControl control = new JobControl("apache-log-group");
            ControlledJob job1 = new ControlledJob(conf);
            job1.setJob(analysis);
            job1.setJobName("analysis");
            ControlledJob job2 = new ControlledJob(conf);
            job2.setJob(counter);
            job2.setJobName("counter");
            job2.addDependingJob(job1);
            control.addJob(job1);
            control.addJob(job2);
            Thread thread = new Thread(control);
            thread.start();
            while (true) {
                if (control.allFinished()) {
                    System.out.println(control.getSuccessfulJobList());
                    control.stop();
                    System.exit(0);
                } else if (control.getFailedJobList().size() > 0) {
                    System.out.println(control.getFailedJobList());
                    control.stop();
                    System.exit(1);
                }
            }
        }else {
            try {
                String tableName = "access-info";
                if(args.length==1){
                    tableName = args[0];
                }
                Scan scan = new Scan();
                scan.setMaxVersions(10);
                Connection connection = ConnectionFactory.createConnection(conf);
                HTable table = (HTable) connection.getTable(TableName.valueOf(tableName));
                ResultScanner rsc = table.getScanner(scan);
                for (Result result : rsc) {
                    //rowKey
                    System.out.println("- "+Bytes.toString(result.getRow()));
                    NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> map = result.getMap();
                    for (byte[] key : map.keySet()) {
                        //columnFamilies
                        System.out.println("   | " + Bytes.toString(key));
                        for (byte[] tk : map.get(key).keySet()) {
                            //columns
                            System.out.println("      |-- " + Bytes.toString(tk));
                            for (Long l : map.get(key).get(tk).keySet()) {
                                //values and timestamps
                                System.out.println("        |-- value[" + Bytes.toString(map.get(key).get(tk).get(l)) + "] (timestamp:" + l + ")");
                            }
                        }
                    }
                }
                rsc.close();
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            }
        }
 }

在主方法中加入了扫描HTable的程序片，完全出于调试的目的，当然可以将其单独写一个方法从Main方法中拿出来，但记得第二个mapreduce任务mapper和reducer输入输出类型不一致，必须设置MapOutPutKey/ValueClass。还可以对输出的序列化文件进行压缩，设置压缩类型和算法，例如按记录压缩，使用Gzip压缩算法：

SequenceFileOutputFormat.setOutputCompressionType(analysis, SequenceFile.CompressionType.RECORD);
SequenceFileOutputFormat.setOutputCompressorClass(analysis, GzipCodec.class);

运行测试

打包并上传到hadoop集群运行（记得HBase得有）：

在RM管理页查看Application的运行：

在HBase Shell中查看表数据记录数：

运行表扫描功能，将输出重定向到文件，运行结束后打开文件查看有层次结构的数据：

最后：在实际环境中，数据量很大的情况下应该注意热点写的问题，使用RowKey散列和预分区。

附

maven项目依赖：

pom.xml

<properties>
    <hadoop.version>2.6.4</hadoop.version>
    <hbase.version>1.2.2</hbase.version>
</properties>
<dependencies>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.11</version>
    </dependency>
    <!--HBase-->
    <dependency>
        <groupId>commons-io</groupId>
        <artifactId>commons-io</artifactId>
        <version>2.4</version>
    </dependency>
    <dependency>
        <groupId>commons-logging</groupId>
        <artifactId>commons-logging</artifactId>
        <version>1.1.3</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-client</artifactId>
        <version>${hbase.version}</version>
    </dependency>
    <dependency>
        <groupId>com.google.protobuf</groupId>
        <artifactId>protobuf-java</artifactId>
        <version>2.5.0</version>
    </dependency>
    <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty</artifactId>
        <version>3.7.0.Final</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-common</artifactId>
        <version>${hbase.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-protocol</artifactId>
        <version>${hbase.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.4.6</version>
    </dependency>
    <dependency>
        <groupId>org.cloudera.htrace</groupId>
        <artifactId>htrace-core</artifactId>
        <version>2.04</version>
    </dependency>
    <dependency>
        <groupId>org.codehaus.jackson</groupId>
        <artifactId>jackson-mapper-asl</artifactId>
        <version>1.9.13</version>
    </dependency>
    <dependency>
        <groupId>org.codehaus.jackson</groupId>
        <artifactId>jackson-core-asl</artifactId>
        <version>1.9.13</version>
    </dependency>
    <dependency>
        <groupId>org.codehaus.jackson</groupId>
        <artifactId>jackson-jaxrs</artifactId>
        <version>1.9.13</version>
    </dependency>
    <dependency>
        <groupId>org.codehaus.jackson</groupId>
        <artifactId>jackson-xc</artifactId>
        <version>1.9.13</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.6</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.5</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-server</artifactId>
        <version>1.2.2</version>
        <exclusions>
            <exclusion>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-hdfs</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
    <!--Hadoop-->
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>${hadoop.version}</version>
    </dependency>
    <dependency>
        <groupId>commons-configuration</groupId>
        <artifactId>commons-configuration</artifactId>
        <version>1.6</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-auth</artifactId>
        <version>${hadoop.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>${hadoop.version}</version>
    </dependency>
    <!-- Protostuff高效序列化 -->
    <dependency>
        <groupId>com.dyuproject.protostuff</groupId>
        <artifactId>protostuff-core</artifactId>
        <version>1.0.8</version>
    </dependency>
    <dependency>
        <groupId>com.dyuproject.protostuff</groupId>
        <artifactId>protostuff-runtime</artifactId>
        <version>1.0.8</version>
    </dependency>
</dependencies>