走向云计算之Hadoop实际应用网站日志分析

一、概述

网站日志分析是Hadoop应用的一个方向。那么什么是网站日志呢?
网站日志是记录web服务器接收处理请求以及运行时错误等各种原始信息的以.log结尾文件。通过网站日志可以清楚的得知用户在什么IP、什么时间、用什么操作系统、什么浏览器、什么分辨率显示器的情况下访问了你网站的哪个页面，是否访问成功。对于SEO优化人员来说，日志就是我们网站优化的指南针，就是我们优化的晴雨表，能够通过分析日志能够有效的修正我们网站优化的方向，从而加快SEO优化的进程，说明日志对于SEO优化人员来说是至关重要的，那么日志我们应该如何分析呢?
生成网站日志一般采用第三方工具，不同的工具产生的日志格式并不相同，这里以apache common日志为例进行分析。
下面我来详细的叙述

27.19.74.143 - - [30/May/2013:17:38:20 +0800] "GET /static/image/common/faq.gif HTTP/1.1" 200 1127

这段网站日志都可以告诉我们哪些有用的信息。

• 27.19.74.143
  这是用户的IP地址，相信大家都能轻易看懂。知道了用户的IP，你甚至可以通过查询来得知用户是来自哪个国家、哪个省份、哪个城市的。查询得知，这个IP来自国内北京地区。不同的网站其用户群会有比较明显的区别。用户IP配合下面的关键词等信息可以让你更加有效的分析网站的用户体验做得够不够好。
• [30/May/2013:17:38:20 +0800]
  这里讲的是发生该处理请求的具体时间。这里的时间是2013年5月30日17点38分20秒。
• GET /static/image/common/faq.gif
  这是服务器的处理动作，一共只有两种：GET和POST。在网站日志中绝大部分都是GET，只有在进行CGI处理的时候才会出现POST，否则绝大多数时间服务器的响应都是GET，也就是用户从服务器上获取了页面或者别的文件。注意了，GET后面还有一个“/”，这里代表的是用户访问的页面，只有一个斜杠自然代表访问的是网站首页。
• HTTP/1.1
  这个代表用户访问该页面的时候，是通过HTTP1.1协议进行传输的，也就是超文本传输1.1版本协议。这个我们不必理会，因为网站日志中除了你可能用FTP之外，普通用户基本都是通过HTTP协议来进行访问的(个别提供FTP下载的网站除外)。
• 200
  代表的是用户访问页面的时候返回的状态码。通常状态码有以下几种：200,301,302,304,404,500等。200代表用户成功的获取到了所请求的文件，如果是搜索引擎，则证明蜘蛛在这次爬行中顺利的发现了一些新的内容。而301则代表用户所访问的某个页面url已经做了301重定向(永久性)处理，302则是暂时性重定向，如果你的网站日志中有过多的302代码，那么你需要注意了，是不是把301做错了做成了302?赶紧修改，搜索引擎不喜欢302重定向。404则代表所访问的页面已经不存在了，或者说访问的url根本就是个错误的。500则是服务器的错误。
• 1127
  这个数字是衡量本次请求数据传输量的大小。

二、网站关键指标

1、浏览量PV

（1）定义：页面浏览量即为PV(Page View)，是指所有用户浏览页面的总和，一个独立用户每打开一个页面就被记录1 次。
（2）分析：网站总浏览量，可以考核用户对于网站的兴趣，就像收视率对于电视剧一样。但是对于网站运营者来说，更重要的是，每个栏目下的浏览量。
该指标计算公式：记录计数，从日志中获取访问次数，又可以细分为各个栏目下的访问次数。

2、注册用户数

每天的注册用户数也是衡量一个网站活力的关键指标，对于本实践来说，用户注册页面为member.php，所以当用户点击注册时请求URL是member.php?mod=register&xxxx。例如：

111.59.57.107 - - [30/May/2013:18:00:40 +0800] "GET /member.php?mod=register&inajax=1 HTTP/1.1" 200 635

该指标计算公式：对访问包含”member.php?mod=register”的url进行计数即可。

3、IP数

（1）定义：一天之内，访问网站的不同独立 IP个数的和。其中同一IP无论访问了几个页面，独立IP数均为1。
（2）分析：从某种程度上来说，独立IP的多少，是衡量网站推广活动好坏最直接的数据。
该指标计算公式：对不同的访问者ip进行计数即可。

4、跳出率

（1）定义：只浏览了一个页面便离开了网站的访问次数占总的访问次数的百分比，即只浏览了一个页面的访问次数 / 全部的访问次数汇总。
（2）分析：跳出率是非常重要的访客黏性指标，它显示了访客对网站的兴趣程度：跳出率越低说明流量质量越好，访客对网站的内容越感兴趣，这些访客越可能是网站的有效用户、忠实用户。该指标也可以衡量网络营销的效果，指出有多少访客被网络营销吸引到宣传产品页或网站上之后，又流失掉了，可以说就是煮熟的鸭子飞了。比如，网站在某媒体上打广告推广，分析从这个推广来源进入的访客指标，其跳出率可以反映出选择这个媒体是否合适，广告语的撰写是否优秀，以及网站入口页的设计是否用户体验良好。
该指标计算公式：先统计一天内只出现一条记录的ip，称为跳出数；然后利用公式 - 跳出数/PV计算得到跳出率。

三、日志采集和分析流程

1、日志数据来源

互联网行业：网站、app、系统（交易系统。。）
传统行业：电信，人们的上网、打电话、发短信等等数据
数据源：网站、app
每次访问都要往后台去发送请求，获取数据，执行业务逻辑；app获取要展现的商品数据；发送请求到后台进行交易和结账。网站/app会发送请求到后台服务器，通常会由Nginx接收请求，并进行转发。

2、后台生成日志

后台服务器，比如Tomcat、Jetty；其实在面向大量用户，高并发（每秒访问量过万）的情况下，通常都不会直接是用Tomcat来接收请求。这种时候，通常，都是用Nginx来接收请求，并且后端接入Tomcat集群/Jetty集群，来进行高并发访问下的负载均衡。比如说，Nginx，或者是Tomcat，你进行适当配置之后，所有请求的数据都会作为log存储起来；接收请求的后台系统（J2EE、PHP、Ruby On Rails），也可以按照你的规范，每接收一个请求，或者每执行一个业务逻辑，就往日志文件里面打一条log。
日志文件（通常由我们预先设定的特殊的格式）通常每天一份。注意一天结束后可能有多份日志文件，因为有多个web服务器，每个服务器都会产生一个日志。

3、日志转移

此时我们需要一个日志转移的工具，比如自己用linux的crontab定时调度一个shell脚本/python脚本；或者自己用java开发一个后台服务，用quartz这样的框架进行定时调度。这个工具，负责将当天的所有日志的数据，都给采集起来，进行合并和处理，等操作；然后作为一份日志文件，给转移到flume agent正在监控的目录中。

4、Flume上传HDFS

flume agent启动起来以后，可以实时的监控linux系统上面的某一个目录，看其中是否有新的文件进来。只要发现有新的日志文件进来，那么flume就会走后续的channel和sink。通常来说，sink都会配置为HDFS。flume负责将每天的一份log文件，传输到HDFS上。

5、HDFS存储日志

HDFS用来存储每天的log数据。为什么用hadoop进行存储呢。因为Hadoop可以存储大数据，大量数据。比如说，每天的日志，数据文件是一个T，那么，也许一天的日志文件，是可以存储在某个Linux系统上面，但是问题是，1个月的呢，1年的呢。当积累了大量数据以后，就不可能存储在单机上，只能存储在Hadoop分布式存储系统中。

6、数据清洗

Hadoop HDFS中的原始的日志数据，会经过数据清洗。为什么要进行数据清洗？因为我们的数据中可能有很多是不符合预期的脏数据。
数据清洗可以使用Hadoop MapReduce，自己开发MR作业，可以用crontab定时调度工具来定时每天执行一次；也可以用Oozie来进行定时调度；也可以（百度、阿里、腾讯、京东、美团）自己组建团队来研发复杂、大型、分布式的调度系统，来承担全公司所有MapReduce / Hive作业的调度（对于大型公司来说，可能每天除了负责数据清洗的MR作业以外，后续的建立数据仓库、进行数据分析和统计的Hive ETL作业可能高达上万个，上十万、百万个），针对HDFS里的原始日志进行数据清洗，写入HDFS中另外一个文件。

7、数据导入Hive中

把HDFS中的清洗后的数据，给导入到Hive的某个表中。这里可以使用动态分区，Hive使用分区表，每个分区放一天的数据。
Hive，底层也是基于HDFS，作为一个大数据的数据仓库。数据仓库内部，再往后，其实就是一些数据仓库建模的ETL。ETL会将原始日志所在的一个表，给转换成几十张，甚至上百张表。这几十，甚至上百张表，就是我们的数据仓库。然后呢，公司的统计分析人员，就会针对数据仓库中的表，执行临时的，或者每天定时调度的Hive SQL ETL作业。来进行大数据的统计和分析。

8、后续统计分析

Spark/Hdoop/Storm可能都会使用Hive中的数据仓库内部的表。此次实践并未使用这一步骤。

9、Sqoop导出数据

为了便于查询和展示，我们通常需要把统计完成的数据存入到Mysql等关系型数据库中，以便网站分析人员的查询和获取。

四、实践详解

1、需要用到的框架

• Flume：用于把网站日志文件上传至HDFS。
• HDFS：用于海量日志文件的存储。
• MapReduce：用于对日志文件进行初步整理。
• Hive：用于对整理后的日志文件作统计分析。
• Sqoop：用于将Hive统计后的结果导入到mysql等关系型数据库中。

注意：启动集群后首先利用命令ntpdate time.nist.gov同步集群时间，否则后面可能出现莫名奇妙的错误。

2、需要的数据

这里给出了两个网站日志文件，文件以上传至云盘。点我。部分日志内容如下：

27.19.74.143 - - [30/May/2013:17:38:20 +0800] "GET /static/image/common/faq.gif HTTP/1.1" 200 1127
110.52.250.126 - - [30/May/2013:17:38:20 +0800] "GET /data/cache/style_1_widthauto.css?y7a HTTP/1.1" 200 1292
27.19.74.143 - - [30/May/2013:17:38:20 +0800] "GET /static/image/common/hot_1.gif HTTP/1.1" 200 680
27.19.74.143 - - [30/May/2013:17:38:20 +0800] "GET /static/image/common/hot_2.gif HTTP/1.1" 200 682

3、通过Flume上传日志文件

把日志数据上传到HDFS中进行处理，可以分为以下几种情况：
（1）如果是日志服务器数据较小、压力较小，可以直接使用shell命令把数据上传到HDFS中；
（2）如果是日志服务器数据较大、压力较大，使用NFS在另一台服务器上上传数据；
（3）如果日志服务器非常多、数据量大，使用flume进行数据处理。
这里虽然只提供了两份日志文件，但考虑到实际使用中多用Flume进行日志上传，这里也用Flume作为示例。

• 这里首先创建两个文件夹：

本地文件夹：mkdir /usr/local/logs
这是Flume监视的文件夹，我们需要把日志文件上传至该文件夹下。
HDFS文件夹：hdfs dfs -mkdir /logs
这是Flume上传文件所在的文件夹，也是日志文件在HDFS的存储位置。

• 改写Flume配置文件

agent1.sources=source1
agent1.sinks=sink1
agent1.channels=channel1

# For each one of the sources, the type is defined
agent1.sources.source1.type=spooldir
agent1.sources.source1.spoolDir=/usr/local/logs
agent1.sources.source1.channels=channel1
agent1.sources.source1.fileHeader = false
agent1.sources.source1.interceptors = i1
agent1.sources.source1.interceptors.i1.type = timestamp
# Each sink's type must be defined
agent1.sinks.sink1.type=hdfs
agent1.sinks.sink1.hdfs.path=hdfs://sparkproject1:9000/logs
agent1.sinks.sink1.hdfs.fileType=DataStream
agent1.sinks.sink1.hdfs.writeFormat=TEXT
agent1.sinks.sink1.hdfs.minBlockReplicas=1  
agent1.sinks.sink1.hdfs.rollInterval=600
agent1.sinks.sink1.hdfs.rollSize=0  
agent1.sinks.sink1.hdfs.rollCount=0  
agent1.sinks.sink1.hdfs.idleTimeout=0
agent1.sinks.sink1.channel=channel1
agent1.sinks.sink1.hdfs.filePrefix=%Y-%m-%d

# Each channel's type is defined.
agent1.channels.channel1.type=file
agent1.channels.channel1.checkpointDir=/usr/local/logs_tmp_cp
agent1.channels.channel1.dataDirs=/usr/local/logs_tmp

• 启动Flume

使用如下命令启动Flume：

flume-ng agent -n agent1 -c conf -f /usr/local/flume/conf/flume-conf.properties -Dflume.root.logger=DEBUG,console

• 上传日志文件

这里使用access_2013_05_30.log这个日志文件来作示例，其在linux上存放的位置是/usr/local/upload目录下。

现在我们将其转移到/usr/local/logs目录下即可通过Flume上传该日志。使用如下命令：

mv access_2013_05_30.log /usr/local/logs/

Flume便自动开始上传该日志。

上传完毕后在HDFS上查看该日志文件：

4、MapReduce进行数据清洗

• 部分日志文件内容

27.19.74.143 - - [30/May/2013:17:38:20 +0800] "GET /static/image/common/faq.gif HTTP/1.1" 200 1127
110.52.250.126 - - [30/May/2013:17:38:20 +0800] "GET /data/cache/style_1_widthauto.css?y7a HTTP/1.1" 200 1292
27.19.74.143 - - [30/May/2013:17:38:20 +0800] "GET /static/image/common/hot_1.gif HTTP/1.1" 200 680
27.19.74.143 - - [30/May/2013:17:38:20 +0800] "GET /static/image/common/hot_2.gif HTTP/1.1" 200 682
27.19.74.143 - - [30/May/2013:17:38:20 +0800] "GET /static/image/filetype/common.gif HTTP/1.1" 200 90
110.52.250.126 - - [30/May/2013:17:38:20 +0800] "GET /source/plugin/wsh_wx/img/wsh_zk.css HTTP/1.1" 200 1482

• 要清理的数据
  （1）我们所要统计分析的均不涉及到访问状态（HTTP状态码）以及本次访问的流量，于是我们首先可以将这两项记录清理掉；
  （2）根据日志记录的数据格式，我们需要将日期格式转换为平常所见的普通格式如20150426这种，于是我们可以写一个类将日志记录的日期进行转换；
  （3）由于静态资源的访问请求对我们的数据分析没有意义，于是我们可以将”GET /staticsource/”开头的访问记录过滤掉，又因为GET和POST字符串对我们也没有意义，因此也可以将其省略掉
• 编写MapReduce程序清理日志
  （1）编写日志解析类对每行记录的五个组成部分进行单独的解析

static class LogParser {
public static final SimpleDateFormat FORMAT = new SimpleDateFormat(
"d/MMM/yyyy:HH:mm:ss", Locale.ENGLISH);
public static final SimpleDateFormat dateformat1 = new SimpleDateFormat(
"yyyyMMddHHmmss");/**
 * 解析英文时间字符串
 * 
 * @param string
 * @return
 * @throws ParseException
 */
private Date parseDateFormat(String string) {
            Date parse = null;
try {
                parse = FORMAT.parse(string);
            } catch (ParseException e) {
                e.printStackTrace();
            }
return parse;
        }

/**
 * 解析日志的行记录
 * 
 * @param line
 * @return 数组含有5个元素，分别是ip、时间、url、状态、流量
 */
public String[] parse(String line) {
            String ip = parseIP(line);
            String time = parseTime(line);
            String url = parseURL(line);
            String status = parseStatus(line);
            String traffic = parseTraffic(line);

return new String[] { ip, time, url, status, traffic };
        }

private String parseTraffic(String line) {
final String trim = line.substring(line.lastIndexOf("\"") + 1)
                    .trim();
            String traffic = trim.split(" ")[1];
return traffic;
        }

private String parseStatus(String line) {
final String trim = line.substring(line.lastIndexOf("\"") + 1)
                    .trim();
            String status = trim.split(" ")[0];
return status;
        }

private String parseURL(String line) {
final int first = line.indexOf("\"");
final int last = line.lastIndexOf("\"");
            String url = line.substring(first + 1, last);
return url;
        }

private String parseTime(String line) {
final int first = line.indexOf("[");
final int last = line.indexOf("+0800]");
            String time = line.substring(first + 1, last).trim();
            Date date = parseDateFormat(time);
return dateformat1.format(date);
        }

private String parseIP(String line) {
            String ip = line.split("- -")[0].trim();
return ip;
        }
    }

（2）编写MapReduce程序对指定日志文件的所有记录进行过滤
Mapper类：

static class MyMapper extends
Mapper<LongWritable, Text, LongWritable, Text> {
        LogParser logParser = new LogParser();
        Text outputValue = new Text();

protected void map(
                LongWritable key,
                Text value,
                org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, LongWritable, Text>.Context context)
                throws java.io.IOException, InterruptedException {
final String[] parsed = logParser.parse(value.toString());

// step1.过滤掉静态资源访问请求
if (parsed[2].startsWith("GET /static/")
                    || parsed[2].startsWith("GET /uc_server")) {
return;
            }
// step2.过滤掉开头的指定字符串
if (parsed[2].startsWith("GET /")) {
                parsed[2] = parsed[2].substring("GET /".length());
            } else if (parsed[2].startsWith("POST /")) {
                parsed[2] = parsed[2].substring("POST /".length());
            }
// step3.过滤掉结尾的特定字符串
if (parsed[2].endsWith(" HTTP/1.1")) {
                parsed[2] = parsed[2].substring(0, parsed[2].length()
                        - " HTTP/1.1".length());
            }
// step4.只写入前三个记录类型项
            outputValue.set(parsed[0] + "\t" + parsed[1] + "\t" + parsed[2]);
            context.write(key, outputValue);
        }
    }

Reducer类：

static class MyReducer extends
Reducer<LongWritable, Text, Text, NullWritable> {
protected void reduce(
                LongWritable k2,
                java.lang.Iterable<Text> v2s,
                org.apache.hadoop.mapreduce.Reducer<LongWritable, Text, Text, NullWritable>.Context context)
                throws java.io.IOException, InterruptedException {
for (Text v2 : v2s) {
                context.write(v2, NullWritable.get());
            }
        };
    }

完整代码如下：

package com.kang.hadoop;

import java.net.URI;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;

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.Reducer;
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;

public class LogCleanJob extends Configured implements Tool {

public static String[] paths={"hdfs://sparkproject1:9000/logs/","hdfs://sparkproject1:9000/logs_out/2013_05_30/"};
public static void main(String[] args) {
        Configuration conf = new Configuration();
try {
int res = ToolRunner.run(conf, new LogCleanJob(), paths);
            System.exit(res);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

@Override
public int run(String[] args) throws Exception {
final Job job = new Job(new Configuration(),
                LogCleanJob.class.getSimpleName());
// 设置为可以打包运行
        job.setJarByClass(LogCleanJob.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 清理已存在的输出文件
        FileSystem fs = FileSystem.get(new URI(args[0]), getConf());
        Path outPath = new Path(args[1]);
if (fs.exists(outPath)) {
            fs.delete(outPath, true);
        }

boolean success = job.waitForCompletion(true);
if(success){
            System.out.println("Clean process success!");
        }
else{
            System.out.println("Clean process failed!");
        }
return 0;
    }

static class MyMapper extends
            Mapper<LongWritable, Text, LongWritable, Text> {
        LogParser logParser = new LogParser();
        Text outputValue = new Text();

protected void map(
                LongWritable key,
                Text value,
                org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, LongWritable, Text>.Context context)
throws java.io.IOException, InterruptedException {
final String[] parsed = logParser.parse(value.toString());

// step1.过滤掉静态资源访问请求
if (parsed[2].startsWith("GET /static/")
                    || parsed[2].startsWith("GET /uc_server")) {
return;
            }
// step2.过滤掉开头的指定字符串
if (parsed[2].startsWith("GET /")) {
                parsed[2] = parsed[2].substring("GET /".length());
            } else if (parsed[2].startsWith("POST /")) {
                parsed[2] = parsed[2].substring("POST /".length());
            }
// step3.过滤掉结尾的特定字符串
if (parsed[2].endsWith(" HTTP/1.1")) {
                parsed[2] = parsed[2].substring(0, parsed[2].length()
                        - " HTTP/1.1".length());
            }
// step4.只写入前三个记录类型项
            outputValue.set(parsed[0] + "\t" + parsed[1] + "\t" + parsed[2]);
            context.write(key, outputValue);
        }
    }

static class MyReducer extends
            Reducer<LongWritable, Text, Text, NullWritable> {
protected void reduce(
                LongWritable k2,
                java.lang.Iterable<Text> v2s,
                org.apache.hadoop.mapreduce.Reducer<LongWritable, Text, Text, NullWritable>.Context context)
throws java.io.IOException, InterruptedException {
for (Text v2 : v2s) {
                context.write(v2, NullWritable.get());
            }
        };
    }

/*
 * 日志解析类
 */
static class LogParser {
public static final SimpleDateFormat FORMAT = new SimpleDateFormat(
"d/MMM/yyyy:HH:mm:ss", Locale.ENGLISH);
public static final SimpleDateFormat dateformat1 = new SimpleDateFormat(
"yyyyMMddHHmmss");

public static void main(String[] args) throws ParseException {
final String S1 = "27.19.74.143 - - [30/May/2013:17:38:20 +0800] \"GET /static/image/common/faq.gif HTTP/1.1\" 200 1127";
            LogParser parser = new LogParser();
final String[] array = parser.parse(S1);
            System.out.println("样例数据： " + S1);
            System.out.format(
"解析结果： ip=%s, time=%s, url=%s, status=%s, traffic=%s",
                    array[0], array[1], array[2], array[3], array[4]);
        }

/**
 * 解析英文时间字符串
 * 
 * @param string
 * @return
 * @throws ParseException
 */
private Date parseDateFormat(String string) {
            Date parse = null;
try {
                parse = FORMAT.parse(string);
            } catch (ParseException e) {
                e.printStackTrace();
            }
return parse;
        }

/**
 * 解析日志的行记录
 * 
 * @param line
 * @return 数组含有5个元素，分别是ip、时间、url、状态、流量
 */
public String[] parse(String line) {
            String ip = parseIP(line);
            String time = parseTime(line);
            String url = parseURL(line);
            String status = parseStatus(line);
            String traffic = parseTraffic(line);

return new String[] { ip, time, url, status, traffic };
        }

private String parseTraffic(String line) {
final String trim = line.substring(line.lastIndexOf("\"") + 1)
                    .trim();
            String traffic = trim.split(" ")[1];
return traffic;
        }

private String parseStatus(String line) {
final String trim = line.substring(line.lastIndexOf("\"") + 1)
                    .trim();
            String status = trim.split(" ")[0];
return status;
        }

private String parseURL(String line) {
final int first = line.indexOf("\"");
final int last = line.lastIndexOf("\"");
            String url = line.substring(first + 1, last);
return url;
        }

private String parseTime(String line) {
final int first = line.indexOf("[");
final int last = line.indexOf("+0800]");
            String time = line.substring(first + 1, last).trim();
            Date date = parseDateFormat(time);
return dateformat1.format(date);
        }

private String parseIP(String line) {
            String ip = line.split("- -")[0].trim();
return ip;
        }
    }
}

运行上述代码，查看结果：

5、使用Hive进行数据统计

• 建立分区表

为了能够借助Hive进行统计分析，首先我们需要将清洗后的数据存入Hive中，那么我们需要先建立一张表。这里我们选择分区表，以日志的当天日期作为分区的指标，建表语句如下：（这里关键之处就在于确定映射的HDFS位置，我这里是/logs_out/,即清洗后的数据存放的位置）。

CREATE EXTERNAL TABLE techbbs(ip string, atime string, url string) PARTITIONED BY (logdate string) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LOCATION '/logs_out';

建立了分区表之后，就需要增加一个分区，增加分区的语句如下：（这里主要针对20130530这一天的日志进行分区）

ALTER TABLE techbbs ADD PARTITION(logdate='2013_05_30') LOCATION '/logs_out/2013_05_30/';

• 使用HQL统计关键指标

（1）关键指标之一：PV量
页面浏览量即为PV(Page View)，是指所有用户浏览页面的总和，一个独立用户每打开一个页面就被记录1 次。这里，我们只需要统计日志中的记录个数即可，HQL代码如下：

CREATE TABLE techbbs_pv_2013_05_30 AS SELECT COUNT(1) AS PV FROM techbbs WHERE logdate='2013_05_30';

查询结果：

（2）关键指标之二：注册用户数
该论坛的用户注册页面为member.php，而当用户点击注册时请求的又是member.php?mod=register的url。因此，这里我们只需要统计出日志中访问的URL是member.php?mod=register的即可，HQL代码如下：

CREATE TABLE techbbs_reguser_2013_05_30 AS SELECT COUNT(1) AS REGUSER FROM techbbs WHERE logdate='2013_05_30' AND INSTR(url,'member.php?mod=register')>0;

查询结果：

（3）关键指标之三：独立IP数
一天之内，访问网站的不同独立 IP 个数加和。其中同一IP无论访问了几个页面，独立IP 数均为1。因此，这里我们只需要统计日志中处理的独立IP数即可，在SQL中我们可以通过DISTINCT关键字，在HQL中也是通过这个关键字：

CREATE TABLE techbbs_ip_2013_05_30 AS SELECT COUNT(DISTINCT ip) AS IP FROM techbbs WHERE logdate='2013_05_30';

查询结果：

（4）关键指标之四：跳出用户数
只浏览了一个页面便离开了网站的访问次数，即只浏览了一个页面便不再访问的访问次数。这里，我们可以通过用户的IP进行分组，如果分组后的记录数只有一条，那么即为跳出用户。将这些用户的数量相加，就得出了跳出用户数，HQL代码如下：

CREATE TABLE techbbs_jumper_2013_05_30 AS SELECT COUNT(1) AS jumper FROM (SELECT COUNT(ip) AS times FROM techbbs WHERE logdate='2013_05_30' GROUP BY ip HAVING times=1) e;

查询结果：

跳出率是指只浏览了一个页面便离开了网站的访问次数占总的访问次数的百分比，即只浏览了一个页面的访问次数 / 全部的访问次数汇总。这里，我们可以将这里得出的跳出用户数/PV数即可得到跳出率。

（5）数据汇总
为了方便通过Sqoop统一导出到MySQL，这里我们借助一张汇总表将刚刚统计到的结果整合起来，通过表连接结合，HQL代码如下：

CREATE TABLE techbbs_2013_05_30 AS SELECT '2013_05_30', a.pv, b.reguser, c.ip, d.jumper FROM techbbs_pv_2013_05_30 a JOIN techbbs_reguser_2013_05_30 b ON 1=1 JOIN techbbs_ip_2013_05_30 c ON 1=1 JOIN techbbs_jumper_2013_05_30 d ON 1=1;

查询结果：

6、使用Sqoop导入到MySQL

• 准备工作：在MySQL中创建结果汇总表

（1）Step1：创建一个新数据库：techbbs

create database techbbs;

（2）Step2：创建一张新数据表：techbbs_logs_stat

CREATE TABLE `techbbs_logs_stat` (
`logdate` varchar(10) NOT NULL,
`pv` int(11) DEFAULT NULL,
`reguser` int(11) DEFAULT NULL,
`ip` int(11) DEFAULT NULL,
`jumper` int(11) DEFAULT NULL,
PRIMARY KEY (`logdate`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8

• 导入操作：通过export命令
  命令如下：

sqoop export --connect jdbc:mysql://sparkproject1:3306/techbbs --username root --password root --table techbbs_logs_stat --fields-terminated-by '\001' --export-dir '/user/hive/warehouse/techbbs_2013_05_30'

其中/user/hive/warehouse/techbbs_2013_05_30是数据表在HDFS上的位置。

查看Mysql数据：

至此，本次日志分析结束。

五、总结

本次日志分析采用Flume进行日志上传，HDFS作为存储，MapReduce进行数据清洗，Hive进行数据统计，Sqoop进行数据导出，Mysql进行结果保存。
你可能会想到我们花费如此长的时间最后得出4个指标，还没有直接本地运算的快。对本次实践来说是这样，因为我们的数据样本也不过60M左右，根本发挥不出Hadoop的优势。在实际应用中，样本数肯定比这个大得多，可能达到1TB/天的数据量，此时使用hadoop来进行存储和分析再合适不过了。用Hadoop的好处是扩展。如果你的数据是一个数TB的单表，那么全表扫描是Hadoop的强项。此外的话（如果你没有这样大数据量的表），请关爱生命，尽量远离Hadoop。它带来的烦恼根本不值，用传统方法既省时又省力。