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 這類所有行都是有效數據的文件。