基於hive的日誌分析系統

轉自 http://www.cppblog.com/koson/archive/2010/07/19/120773.html        

  hive 簡介

        hive 是一個基於 hadoop 的開源數據倉庫工具，用於存儲和處理海量結構化數據。    它把海量數據存儲於 hadoop 文件系統，而不是數據庫，但提供了一套類數據庫的數據存儲和處理機制，並採用 HQL （類 SQL ）語言對這些數據進行自動化管理和處理。咱們能夠把 hive 中海量結構化數據當作一個個的表，而實際上這些數據是分佈式存儲在 HDFS 中的。 Hive 通過對語句進行解析和轉換，最終生成一系列基於 hadoop 的 map/reduce 任務，經過執行這些任務完成數據處理。

        Hive 誕生於 facebook 的日誌分析需求，面對海量的結構化數據， hive 以較低的成本完成了以往須要大規模數據庫才能完成的任務，而且學習門檻相對較低，應用開發靈活而高效。

        Hive 自 2009.4.29 發佈第一個官方穩定版 0.3.0 至今，不過一年的時間，正在慢慢完善，網上能找到的相關資料至關少，尤爲中文資料更少，本文結合業務對 hive 的應用作了一些探索，並把這些經驗作一個總結，所謂前車可鑑，但願讀者能少走一些彎路。

        Hive 的官方 wiki 請參考這裏 :

        http://wiki.apache.org/hadoop/Hive

        官方主頁在這裏：

        http://hadoop.apache.org/hive/

        hive-0.5.0 源碼包和二進制發佈包的下載地址

        http://labs.renren.com/apache-mirror/hadoop/hive/hive-0.5.0/

2、           部署

        因爲 Hive 是基於 hadoop 的工具，因此 hive 的部署須要一個正常運行的 hadoop 環境。如下介紹 hive 的簡單部署和應用。

        部署環境：

        操做系統： Red Hat Enterprise Linux AS release 4 (Nahant Update 7)

        Hadoop ： hadoop-0.20.2 ，正常運行

        部署步驟以下：

一、   下載最新版本發佈包 hive-0.5.0-dev.tar.gz ，傳到 hadoop 的 namenode 節點上，解壓獲得 hive 目錄。假設路徑爲： /opt/hadoop/hive-0.5.0-bin

二、   設置環境變量 HIVE_HOME ，指向 hive 根目錄 /opt/hadoop/hive-0.5.0-bin 。因爲 hadoop 已運行，檢查環境變量 JAVA_HOME 和 HADOOP_HOME 是否正確有效。

三、   切換到 $HIVE_HOME 目錄， hive 配置默認便可，運行 bin/hive 便可啓動 hive ，若是正常啓動，將會出現「 hive> 」提示符。

四、   在命令提示符中輸入「 show tables; 」，若是正常運行，說明已部署成功，可供使用。

常見問題：

一、        執行「 show tables; 」命令提示「 FAILED: Error in metadata: java.lang.IllegalArgumentException: URI:  does not have a scheme 」，這是因爲 hive 找不到存放元數據庫的數據庫而致使的，修改 conf/ hive-default.xml 配置文件中的 hive.metastore.local 爲 true 便可。因爲 hive 把結構化數據的元數據信息放在第三方數據庫，此處設置爲 true ， hive 將在本地建立 derby 數據庫用於存放元數據。固然若是有須要也能夠採用 mysql 等第三方數據庫存放元數據，不過這時 hive.metastore.local 的配置值應爲 false 。

二、        若是你已有一套 nutch1.0 系統正在跑，而你不想單獨再去部署一套 hadoop 環境，你能夠直接使用 nutch1.0 自帶的 hadoop 環境，但這樣的部署會致使 hive 不能正常運行，提示找不到某些方法。這是因爲 nutch1.0 使用了 commons-lang-2.1.jar 這個包，而 hive 須要的是 commons-lang-2.4.jar ，下載一個 2.4 版本的包替換掉 2.1 便可， nutch 和 hive 都能正常運行。

3、           應用場景

        本文主要講述使用 hive 的實踐，業務不是關鍵，簡要介紹業務場景，本次的任務是對搜索日誌數據進行統計分析。

        集團搜索剛上線不久，日誌量並不大 。這些日誌分佈在 5 臺前端機，按小時保存，並以小時爲週期定時將上一小時產生的數據同步到日誌分析機，統計數據要求按小時更新。這些統計項，包括關鍵詞搜索量 pv ，類別訪問量，每秒訪問量 tps 等等。

基於 hive ，咱們將這些數據按天爲單位建表，天天一個表，後臺腳本根據時間戳將每小時同步過來的 5 臺前端機的日誌數據合併成一個日誌文件，導入 hive 系統，每小時同步的日誌數據被追加到當天數據表中，導入完成後，當天各項統計項將被從新計算並輸出統計結果。

        以上需求若直接基於 hadoop 開發，須要自行管理數據，針對多個統計需求開發不一樣的 map/reduce 運算任務，對合並、排序等多項操做進行定製，並檢測任務運行狀態，工做量並不小。但使用 hive ，從導入到分析、排序、去重、結果輸出，這些操做均可以運用 hql 語句來解決，一條語句通過處理被解析成幾個任務來運行，即便是關鍵詞訪問量增量這種須要同時訪問多天數據的較爲複雜的需求也能經過表關聯這樣的語句自動完成，節省了大量工做量。

4、           Hive 實戰

        初次使用 hive ，應該說上手仍是挺快的。 Hive 提供的類 SQL 語句與 mysql 語句極爲類似，語法上有大量相同的地方，這給咱們上手帶來了很大的方便，可是要駕輕就熟地寫好這些語句，還須要對 hive 有較好的瞭解，才能結合 hive 特點寫出精妙的語句。

        關於 hive 語言的詳細語法可參考官方 wiki 的語言手冊 :

        http://wiki.apache.org/hadoop/Hive/LanguageManual

        雖然語法風格爲咱們提供了便利，但初次使用遇到的問題仍是很多的，下面針對業務場景談談咱們遇到的問題，和對 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 方法便可，示例代碼以下：

    public boolean next(LongWritable key, BytesWritable value)

        throws IOException {

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

        String strReplace = text .toString().toLowerCase().replace( "@$_$@" , "\001" );

        Text txtReplace = new Text();

        txtReplace.set(strReplace );

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

        return true ;

      }

         return false ;

}

        重寫 HiveIgnoreKeyTextOutputFormat 中 RecordWriter 中的 write 方法，示例代碼以下：

    public void write (Writable w) throws IOException {

      String strReplace = ((Text)w).toString().replace( "\001" , "@$_$@" );

      Text txtReplace = new Text();

      txtReplace.set(strReplace);

      byte [] output = txtReplace.getBytes();

      bytesWritable .set(output, 0, output. length );

      writer .write( bytesWritable );

}

        自定義 outputformat/inputformat 後，在建表時須要指定 outputformat/inputformat ，以下示例：

stored as INPUTFORMAT 'com.aspire.search.loganalysis.hive.SearchLogInputFormat' OUTPUTFORMAT 'com.aspire.search.loganalysis.hive.SearchLogOutputFormat'

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_20100603 as select keyword,count(keyword) as count from searchlog_20100603 group by keyword order by count desc;

c)          統計每類用戶終端的搜索量，並按搜索量降序排列，而後把結果存入表 device_20100603 中

create table device_20100603 as select device,count(device) as count from searchlog_20100603 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;

j)             

 

四、 用戶自定義函數 UDF

部分統計結果須要以 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;

五、   輸出 XML 格式的統計結果

前面看到部分日誌統計結果輸出到一個表中，藉助 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 這類全部行都是有效數據的文件。

5、           總結

        Hive 是一個可擴展性極強的數據倉庫工具，藉助於 hadoop 分佈式存儲計算平臺和 hive 對 SQL 語句的理解能力，咱們所要作的大部分工做就是輸入和輸出數據的適配，偏偏這兩部分 IO 格式是變幻無窮的，咱們只須要定製咱們本身的輸入輸出適配器， hive將爲咱們透明化存儲和處理這些數據，大大簡化咱們的工做。本文的重心也正在於此，這部分工做相信每個作數據分析的朋友都會面對的，但願對您有益。

        本文介紹了一次至關簡單的基於 hive 的日誌統計實戰，對 hive 的運用還處於一個相對較淺的層面，目前尚能知足需求。對於一些較複雜的數據分析任務，以上所介紹的經驗極可能是不夠用的，甚至是 hive 作不到的， hive 還有不少進階功能，限於篇幅本文未能涉及，待往後結合具體任務再詳細闡述。