Hive实战之日志分析

首先遇到的是日志数据的分隔符问题，我们的日志数据的大致格式如下：

2010-05-24 00:00:02@$_$@QQ2010@$_$@all@$_$@NOKIA_1681C@$_$@1@$_$@10@$_$@@$_$@-1@$_$@10@$_$@application@$_$@1

从 格式可见其分隔符是“ @$_$@ ”，这是为了尽可能防止日志正文出现与分隔符相同的字符而导致数据混淆。本来 hive支持在建表的时候指定自定义分隔符的，但经过多次测试发现只支持单个字符的自定义分隔符，像“ @$_$@ ”这样的分隔符是不能被支持的，但是我们可以通过对分隔符的定制解决这个问题， hive 的内部分隔符是“ \001 ”，只要把分隔符替换成“\001 ”即可。

经过探索我们发现有两条途径解决这个问题。

a)自定义 outputformat 和 inputformat 。

Hive 的 outputformat/inputformat 与 hadoop 的 outputformat/inputformat 相当类似， inputformat 负责把输入数据进行格式化，然后提供给 hive ， outputformat 负责把 hive 输出的数据重新格式化成目标格式再输出到文件，这种对格式进行定制的方式较为底层，对其进行定制也相对简单，重写 InputFormat 中 RecordReader 类中的 next 方法即可，示例代码如下：

package com.rimi.hive.customer;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.FileSplit;

import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapred.LineRecordReader;

import org.apache.hadoop.mapred.RecordReader;

public class RimiRecordReader implements RecordReader<LongWritable, Text> {

 // Reader

 private LineRecordReader reader;

 public RimiRecordReader(JobConf job, FileSplit split) throws IOException {

  reader = new LineRecordReader(job, split);

 }

 @Override

 public void close() throws IOException {

  reader.close();

 }

 @Override

 public boolean next(LongWritable key, Text value) throws IOException {

   while ( reader .next(key, value ) ) {

         String strReplace = value .toString().toLowerCase().replace( "@$_$@" , "\t" );//替换@$_$@为 tab分隔符

         Text txtReplace = new Text();

         txtReplace.set(strReplace );

         value.set(txtReplace.getBytes(), 0, txtReplace.getLength());

         return true ;

       }

     return false ;

 }

 @Override

 public float getProgress() throws IOException {

  return reader.getProgress();

 }

 @Override

 public LongWritable createKey() {

  return new LongWritable(0);

 }

 @Override

 public Text createValue() {

  return new Text("");

 }

 @Override

 public long getPos() throws IOException {

  return reader.getPos();

 }

}

********************************************************

package com.rimi.hive.customer;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.FileSplit;

import org.apache.hadoop.mapred.InputSplit;

import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapred.JobConfigurable;

import org.apache.hadoop.mapred.RecordReader;

import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.mapred.TextInputFormat;

public class RimiInputFormat extends TextInputFormat implements JobConfigurable {

public RecordReader<LongWritable, Text> getRecordReader(

    InputSplit genericSplit, JobConf job, Reporter reporter)

    throws IOException {

  reporter.setStatus(genericSplit.toString());

  return new RimiRecordReader(job,(FileSplit) genericSplit);

 }

}    

需要添加jar包的方式导入资源包：

1. 使用add jar path/test.jar;方法加入 ，每次session有作用

该方法的缺点是每次启动Hive的时候都要从新加入，退出hive就会失效。

2. 通过设置hive的配置文件hive-site.xml 加入

在配置文件中增加配置

<property>

<name>hive.aux.jars.path</name>

<value>file:///jarpath/all_new1.jar,file:///jarpath/all_new2.jar</value>

</property>

保存即可。

该方法比第一种方法方便很多。不需要每次启动Hive执行命令加入，只是配置稍微复杂一些。

3. 在${HIVE_HOME中创建文件夹auxlib ，然后将自定义jar文件放入该文件夹中。

个人推荐这种方法，方便快捷。

创建表：

自定义 outputformat/inputformat 后，在建表时需要指定 outputformat/inputformat ，如下示例：

create table visit (time string, keyword string, device string, ip string) 

ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'

stored as INPUTFORMAT 'com.rimi.hive.customer.RimiInputFormat'

OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'; 

创建有分区的外部表：

create EXTERNAL table visit (time string, keyword string, device string, ip string)

PARTITIONED BY (date string )

ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'

stored as INPUTFORMAT 'com.rimi.hive.customer.RimiInputFormat'

OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'; 

导入数据 

alter table visit add partition (date='2014-05-23') location '/hive_test/rimi_2014-05-23.log';

导入数据：

LOAD DATA LOCAL INPATH '/home/rimi/rimilog/rimi.log' OVERWRITE INTO TABLE visit;

b) 通过 SerDe(serialize/deserialize) ，在数据序列化和反序列化时格式化数据。

这种方式稍微复杂一点，对数据的控制能力也要弱一些，它使用正则表达式来匹配和处理数据，性能也会有所影响。但它的优点是可以自定义表属性信息 SERDEPROPERTIES ，在 SerDe 中通过这些属性信息可以有更多的定制行为。

a) 多版本日志格式的兼容

由于 hive 的应用场景主要是处理冷数据（只读不写），因此它只支持批量导入和导出数据，并不支持单条数据的写入或更新，所以如果要导入的数据存在某些不太规范的行，则需要我们定制一些扩展功能对其进行处理。

我 们需要处理的日志数据存在多个版本，各个版本每个字段的数据内容存在一些差异，可能版本 A 日志数据的第二个列是搜索关键字，但版本 B 的第二列却是搜索的终端类型，如果这两个版本的日志直接导入 hive 中，很明显数据将会混乱，统计结果也不会正确。我们的任务是要使多个版本的日志数据能在 hive 数据仓库中共存，且表的 input/output 操作能够最终映射到正确的日志版本的正确字段。

这里我们不关心这部分繁琐的工作，只关心技术实现的关键点，这个功能该在哪里实现才能让 hive 认得这些不同格式的数据呢？经过多方尝试，在中间任何环节做这个版本适配都将导致复杂化，最终这个工作还是在 inputformat/outputformat 中完成最为优雅，毕竟 inputformat 是源头， outputformat 是最终归宿。具体来说，是在前面提到的 inputformat 的 next 方法中和在 outputformat 的 write 方法中完成这个适配工作。

b) Hive 操作本地数据

一开始，总是把本地数据先传到 HDFS ，再由 hive 操作 hdfs 上的数据，然后再把数据从 HDFS 上传回本地数据。后来发现大可不必如此， hive 语句都提供了“ local ”关键字，支持直接从本地导入数据到 hive ，也能从 hive 直接导出数据到本地，不过其内部计算时当然是用 HDFS 上的数据，只是自动为我们完成导入导出而已。

日志数据的统计处理在这里反倒没有什么特别之处，就是一些 SQL 语句而已，也没有什么高深的技巧，不过还是列举一些语句示例，以示 hive 处理数据的方便之处，并展示 hive 的一些用法。

a) 为 hive 添加用户定制功能，自定义功能都位于 hive_contrib.jar 包中

add jar /opt/hadoop/hive-0.5.0-bin/lib/hive_contrib.jar;

b)  统计每个关键词的搜索量，并按搜索量降序排列，然后把结果存入表 keyword_20100603 中

create table keyword_20140523 as select keyword,count(keyword) as count from visit group by keyword order by count desc; 、

建议创建表分区，此处不用创建外部表，因为表数据不太大

hive>create table keyword (keyword string,count int) partitioned by (date string) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' ;

hive>alter table keyword add partition (date='2014-05-23') ;

hive> INSERT OVERWRITE TABLE keyword PARTITION (date='2014-05-23') select keyword,count(keyword) as count from visit where date='2014-05-23' group by keyword order by count desc;

c) 统计每类用户终端的搜索量，并按搜索量降序排列，然后把结果存入表 device_20100603 中

create table device_20140523 as select device,count(device) as count from visit group by device order by count desc;

hive>create table device (keyword string,count int) partitioned by (date string) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'; 

hive> alter table device add partition (date='2014-05-23');

hive> INSERT OVERWRITE TABLE device PARTITION (date='2014-05-23') select device,count(device) as count from visit where date='2014-05-23' group by device order by count desc

d) 创建表 time_20100603 ，使用自定义的 INPUTFORMAT 和 OUTPUTFORMAT ，并指定表数据的真实存放位置在 '/LogAnalysis/results/time_20100603' （ HDFS 路径），而不是放在 hive 自己的数据目录中

create external table if not exists time_20100603(time string, count int) stored as INPUTFORMAT 'com.aspire.search.loganalysis.hive.XmlResultInputFormat' OUTPUTFORMAT 'com.aspire.search.loganalysis.hive.XmlResultOutputFormat' LOCATION '/LogAnalysis/results/time_20100603';

e) 统计每秒访问量 TPS ，按访问量降序排列，并把结果输出到表 time_20100603 中，这个表我们在上面刚刚定义过，其真实位置在 '/LogAnalysis/results/time_20100603' ，并且由于 XmlResultOutputFormat 的格式化，文件内容是 XML 格式。

insert overwrite table time_20100603 select time,count(time) as count from searchlog_20100603 group by time order by count desc;

f) 计算每个搜索请求响应时间的最大值，最小值和平均值

insert overwrite table response_20100603 select max(responsetime) as max,min(responsetime) as min,avg(responsetime) as avg from searchlog_20100603;

g)创建一个表用于存放今天与昨天的关键词搜索量和增量及其增量比率，表数据位于 '/LogAnalysis/results/keyword_20100604_20100603' ，内容将是 XML 格式。

create external table if not exists keyword_20100604_20100603(keyword string, count int, increment int, incrementrate double) stored as INPUTFORMAT 'com.aspire.search.loganalysis.hive.XmlResultInputFormat' OUTPUTFORMAT 'com.aspire.search.loganalysis.hive.XmlResultOutputFormat' LOCATION '/LogAnalysis/results/keyword_20100604_20100603';

h)设置表的属性，以便 XmlResultInputFormat 和 XmlResultOutputFormat 能根据 output.resulttype 的不同内容输出不同格式的 XML 文件。

alter table keyword_20100604_20100603 set tblproperties ('output.resulttype'='keyword');

i) 关联今天关键词统计结果表（ keyword_20100604 ）与昨天关键词统计结果表（ keyword_20100603 ），统计今天与昨天同时出现的关键词的搜索次数，今天相对昨天的增量和增量比率，并按增量比率降序排列，结果输出到刚刚定义的 keyword_20100604_20100603 表中，其数据文件内容将为 XML 格式。

insert overwrite table keyword_20100604_20100603 select cur.keyword, cur.count, cur.count-yes.count as increment, (cur.count-yes.count)/yes.count as incrementrate from keyword_20100604 cur join keyword_20100603 yes on (cur.keyword = yes.keyword) order by incrementrate desc;

部分统计结果需要以 CSV 的格式输出，对于这类文件体全是有效内容的文件，不需要像 XML 一样包含 version ， encoding 等信息的文件头，最适合用 UDF(user define function) 了。

UDF 函数可直接应用于 select 语句，对查询结构做格式化处理之后，再输出内容。自定义 UDF 需要继承 org.apache.hadoop.hive.ql.exec.UDF ，并实现 evaluate 函数， Evaluate 函数支持重载，还支持可变参数。我们实现了一个支持可变字符串参数的 UDF ，支持把 select 得出的任意个数的不同类型数据转换为字符串后，按 CSV 格式输出，由于代码较简单，这里给出源码示例：

public String evaluate(String... strs) {        StringBuilder sb = new StringBuilder();        for ( int i = 0; i < strs. length ; i++) {            sb.append(ConvertCSVField(strs[i])).append( ',' );        }        sb.deleteCharAt(sb.length()-1);        return sb.toString(); }

需要注意的是，要使用 UDF 功能，除了实现自定义 UDF 外，还需要加入包含 UDF 的包，示例：

add jar /opt/hadoop/hive-0.5.0-bin/lib/hive_contrib.jar;

然后创建临时方法，示例：

CREATE TEMPORARY FUNCTION Result2CSv AS ‘com.aspire.search.loganalysis.hive. Result2CSv';

使用完毕还要 drop 方法，示例：

DROP TEMPORARY FUNCTION Result2CSv;

前 面看到部分日志统计结果输出到一个表中，借助 XmlResultInputFormat 和 XmlResultOutputFormat 格式化成 XML 文件，考虑到创建这个表只是为了得到 XML 格式的输出数据，我们只需实现 XmlResultOutputFormat 即可，如果还要支持 select 查询，则我们还需要实现 XmlResultInputFormat ，这里我们只介绍 XmlResultOutputFormat 。

前面介绍过，定制 XmlResultOutputFormat 我们只需重写 write 即可，这个方法将会把 hive 的以 ’\001’ 分隔的多字段数据格式化为我们需要的 XML 格式，被简化的示例代码如下：

public void write(Writable w) throws IOException {            String[] strFields = ((Text) w).toString().split( "\001" );            StringBuffer sbXml = new StringBuffer();            if ( strResultType .equals( "keyword" )) {     sbXml.append( "<record><keyword>" ).append(strFields[0]).append(     "</keyword><count>" ).append(strFields[1]).append(           "</count><increment>" ).append(strFields[2]).append(     "</increment><rate>" ).append(strFields[3]).append( "</rate></result>" );            }            Text txtXml = new Text();            byte [] strBytes = sbXml.toString().getBytes( "utf-8" );            txtXml.set(strBytes, 0, strBytes. length );            byte [] output = txtXml.getBytes();            bytesWritable .set(output, 0, output. length );            writer .write( bytesWritable );     }

其中的 strResultType .equals( "keyword" ) 指定关键词统计结果，这个属性来自以下语句对结果类型的指定，通过这个属性我们还可以用同一个 outputformat 输出多种类型的结果。

alter table keyword_20100604_20100603 set tblproperties ('output.resulttype'='keyword');

仔 细看看 write 函数的实现便可发现，其实这里只输出了 XML 文件的正文，而 XML 的文件头和结束标签在哪里输出呢？所幸我们采用的是基于 outputformat 的实现，我们可以在构造函数输出 version ， encoding 等文件头信息，在 close() 方法中输出结束标签。

这也是我们为什么不使用 UDF 来输出结果的原因，自定义 UDF 函数不能输出文件头和文件尾，对于 XML 格式的数据无法输出完整格式，只能输出 CSV 这类所有行都是有效数据的文件。