2. 程式人生 > >Hive UDAF開發詳解

Hive UDAF開發詳解

阿新 • • 發佈：2017-07-05
-s 聚合 而且 pri ros cal 關系 方法調用 evaluator

明

這篇文章是來自Hadoop Hive UDAF Tutorial - Extending Hive with Aggregation Functions：的不嚴格翻譯，因為翻譯的文章示例寫得比較通俗易懂，此外，我把自己對於Hive的UDAF理解穿插到文章裏面。

udfa是hive中用戶自定義的聚集函數，hive內置UDAF函數包括有sum()與count（），UDAF實現有簡單與通用兩種方式，簡單UDAF因為使用Java反射導致性能損失，而且有些特性不能使用，已經被棄用了；在這篇博文中我們將關註Hive中自定義聚類函數-GenericUDAF，UDAF開發主要涉及到以下兩個抽象類：

org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver
org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator

源碼鏈接

博文中的所有的代碼和數據可以在以下鏈接找到：hive examples

示例數據準備

首先先創建一張包含示例數據的表：people，該表只有name一列，該列中包含了一個或多個名字，該表數據保存在people.txt文件中。

~$ cat ./people.txt

John Smith
John and Ann White
Ted Green
Dorothy

把該文件上載到hdfs目錄/user/matthew/people中：

hadoop fs -mkdir people
hadoop fs -put ./people.txt people

下面要創建hive外部表，在hive shell中執行

CREATE EXTERNAL TABLE people (name string)
ROW FORMAT DELIMITED FIELDS 
	TERMINATED BY ‘\t‘ 
	ESCAPED BY ‘‘ 
	LINES TERMINATED BY ‘\n‘
STORED AS TEXTFILE 
LOCATION ‘/user/matthew/people‘;

相關抽象類介紹

創建一個GenericUDAF必須先了解以下兩個抽象類：
org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver 
org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator

為了更好理解上述抽象類的API，要記住hive只是mapreduce函數，只不過hive已經幫助我們寫好並隱藏mapreduce，向上提供簡潔的sql函數，所以我們要結合Mapper、Combiner與Reducer來幫助我們理解這個函數。要記住在hadoop集群中有若幹臺機器，在不同的機器上Mapper與Reducer任務獨立運行。

所以大體上來說，這個UDAF函數讀取數據（mapper），聚集一堆mapper輸出到部分聚集結果（combiner），並且最終創建一個最終的聚集結果（reducer）。因為我們跨域多個combiner進行聚集，所以我們需要保存部分聚集結果。

AbstractGenericUDAFResolver

Resolver很簡單，要覆蓋實現下面方法，該方法會根據sql傳人的參數數據格式指定調用哪個Evaluator進行處理。

<span style="background-color: rgb(255, 255, 255);"><span style="font-size:14px;">public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters) throws SemanticException;</span></span>

GenericUDAFEvaluator

UDAF邏輯處理主要發生在Evaluator中，要實現該抽象類的幾個方法。

在理解Evaluator之前，必須先理解objectInspector接口與GenericUDAFEvaluator中的內部類Model。

ObjectInspector

作用主要是解耦數據使用與數據格式，使得數據流在輸入輸出端切換不同的輸入輸出格式，不同的Operator上使用不同的格式。可以參考這兩篇文章：first post on Hive UDFs、Hive中ObjectInspector的作用，裏面有關於objectinspector的介紹。

Model

Model代表了UDAF在mapreduce的各個階段。

public static enum Mode {
    /**
     * PARTIAL1: 這個是mapreduce的map階段:從原始數據到部分數據聚合
     * 將會調用iterate()和terminatePartial()
     */
    PARTIAL1,
        /**
     * PARTIAL2: 這個是mapreduce的map端的Combiner階段，負責在map端合並map的數據::從部分數據聚合到部分數據聚合:
     * 將會調用merge() 和 terminatePartial() 
     */
    PARTIAL2,
        /**
     * FINAL: mapreduce的reduce階段:從部分數據的聚合到完全聚合 
     * 將會調用merge()和terminate()
     */
    FINAL,
        /**
     * COMPLETE: 如果出現了這個階段，表示mapreduce只有map，沒有reduce，所以map端就直接出結果了:從原始數據直接到完全聚合
      * 將會調用 iterate()和terminate()
     */
    COMPLETE
  };

一般情況下，完整的UDAF邏輯是一個mapreduce過程，如果有mapper和reducer，就會經歷PARTIAL1(mapper)，FINAL(reducer)，如果還有combiner，那就會經歷PARTIAL1(mapper)，PARTIAL2(combiner)，FINAL(reducer)。

而有一些情況下的mapreduce，只有mapper，而沒有reducer，所以就會只有COMPLETE階段，這個階段直接輸入原始數據，出結果。

GenericUDAFEvaluator的方法

// 確定各個階段輸入輸出參數的數據格式ObjectInspectors
public  ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException;

// 保存數據聚集結果的類
abstract AggregationBuffer getNewAggregationBuffer() throws HiveException;

// 重置聚集結果
public void reset(AggregationBuffer agg) throws HiveException;

// map階段，叠代處理輸入sql傳過來的列數據
public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException;

// map與combiner結束返回結果，得到部分數據聚集結果
public Object terminatePartial(AggregationBuffer agg) throws HiveException;

// combiner合並map返回的結果，還有reducer合並mapper或combiner返回的結果。
public void merge(AggregationBuffer agg, Object partial) throws HiveException;

// reducer階段，輸出最終結果
public Object terminate(AggregationBuffer agg) throws HiveException;

圖解Model與Evaluator關系

技術分享

Model各階段對應Evaluator方法調用


技術分享

Evaluator各個階段下處理mapreduce流程

實例

下面將講述一個聚集函數UDAF的實例，我們將計算people這張表中的name列字母的個數。

下面的函數代碼是計算指定列中字符的總數（包括空格）

代碼

@Description(name = "letters", value = "_FUNC_(expr) - 返回該列中所有字符串的字符總數")
public class TotalNumOfLettersGenericUDAF extends AbstractGenericUDAFResolver {

    @Override
    public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)
            throws SemanticException {
        if (parameters.length != 1) {
            throw new UDFArgumentTypeException(parameters.length - 1,
                    "Exactly one argument is expected.");
        }
        
        ObjectInspector oi = TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(parameters[0]);
        
        if (oi.getCategory() != ObjectInspector.Category.PRIMITIVE){
            throw new UDFArgumentTypeException(0,
                            "Argument must be PRIMITIVE, but "
                            + oi.getCategory().name()
                            + " was passed.");
        }
        
        PrimitiveObjectInspector inputOI = (PrimitiveObjectInspector) oi;
        
        if (inputOI.getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING){
            throw new UDFArgumentTypeException(0,
                            "Argument must be String, but "
                            + inputOI.getPrimitiveCategory().name()
                            + " was passed.");
        }
        
        return new TotalNumOfLettersEvaluator();
    }

    public static class TotalNumOfLettersEvaluator extends GenericUDAFEvaluator {

        PrimitiveObjectInspector inputOI;
        ObjectInspector outputOI;
        PrimitiveObjectInspector integerOI;
        
        int total = 0;

        @Override
        public ObjectInspector init(Mode m, ObjectInspector[] parameters)
                throws HiveException {
        	
            assert (parameters.length == 1);
            super.init(m, parameters);
           
             //map階段讀取sql列，輸入為String基礎數據格式
            if (m == Mode.PARTIAL1 || m == Mode.COMPLETE) {
                inputOI = (PrimitiveObjectInspector) parameters[0];
            } else {
			//其余階段，輸入為Integer基礎數據格式
            	integerOI = (PrimitiveObjectInspector) parameters[0];
            }

             // 指定各個階段輸出數據格式都為Integer類型
            outputOI = ObjectInspectorFactory.getReflectionObjectInspector(Integer.class,
                    ObjectInspectorOptions.JAVA);
            return outputOI;

        }

        /**
         * 存儲當前字符總數的類
         */
        static class LetterSumAgg implements AggregationBuffer {
            int sum = 0;
            void add(int num){
            	sum += num;
            }
        }

        @Override
        public AggregationBuffer getNewAggregationBuffer() throws HiveException {
            LetterSumAgg result = new LetterSumAgg();
            return result;
        }

        @Override
        public void reset(AggregationBuffer agg) throws HiveException {
        	LetterSumAgg myagg = new LetterSumAgg();
        }
        
        private boolean warned = false;

        @Override
        public void iterate(AggregationBuffer agg, Object[] parameters)
                throws HiveException {
            assert (parameters.length == 1);
            if (parameters[0] != null) {
                LetterSumAgg myagg = (LetterSumAgg) agg;
                Object p1 = ((PrimitiveObjectInspector) inputOI).getPrimitiveJavaObject(parameters[0]);
                myagg.add(String.valueOf(p1).length());
            }
        }

        @Override
        public Object terminatePartial(AggregationBuffer agg) throws HiveException {
            LetterSumAgg myagg = (LetterSumAgg) agg;
            total += myagg.sum;
            return total;
        }

        @Override
        public void merge(AggregationBuffer agg, Object partial)
                throws HiveException {
            if (partial != null) {
                
                LetterSumAgg myagg1 = (LetterSumAgg) agg;
                
                Integer partialSum = (Integer) integerOI.getPrimitiveJavaObject(partial);
                
                LetterSumAgg myagg2 = new LetterSumAgg();
                
                myagg2.add(partialSum);
                myagg1.add(myagg2.sum);
            }
        }

        @Override
        public Object terminate(AggregationBuffer agg) throws HiveException {
            LetterSumAgg myagg = (LetterSumAgg) agg;
            total = myagg.sum;
            return myagg.sum;
        }

    }
}

代碼說明

AggregationBuffer 允許我們保存中間結果，通過定義我們的buffer，我們可以處理任何格式的數據，在代碼例子中字符總數保存在AggregationBuffer 。

/**
* 保存當前字符總數的類
*/
static class LetterSumAgg implements AggregationBuffer {
	int sum = 0;
	void add(int num){
		sum += num;
	}
}

這意味著UDAF在不同的mapreduce階段會接收到不同的輸入。Iterate讀取我們表中的一行（或者準確來說是表），然後輸出其他數據格式的聚集結果。

artialAggregation合並這些聚集結果到另外相同格式的新的聚集結果，然後最終的reducer取得這些聚集結果然後輸出最終結果（該結果或許與接收數據的格式不一致）。

在init()方法中我們指定輸入為string，結果輸出格式為integer，還有，部分聚集結果輸出格式為integer（保存在aggregation buffer中）；terminate()與terminatePartial()兩者輸出一個integer。

// init方法中根據不同的mode指定輸出數據的格式objectinspector
if (m == Mode.PARTIAL1 || m == Mode.COMPLETE) {
	inputOI = (PrimitiveObjectInspector) parameters[0];
} else {
	integerOI = (PrimitiveObjectInspector) parameters[0];
}

// 不同model階段的輸出數據格式
outputOI = ObjectInspectorFactory.getReflectionObjectInspector(Integer.class,
                    ObjectInspectorOptions.JAVA);

iterate()函數讀取到每行中列的字符串，計算與保存該字符串的長度

public void iterate(AggregationBuffer agg, Object[] parameters)
	throws HiveException {
	...
	Object p1 = ((PrimitiveObjectInspector) inputOI).getPrimitiveJavaObject(parameters[0]);
	myagg.add(String.valueOf(p1).length());
	}
}

Merge函數增加部分聚集總數到AggregationBuffer

public void merge(AggregationBuffer agg, Object partial)
      	throws HiveException {
	if (partial != null) {
                
		LetterSumAgg myagg1 = (LetterSumAgg) agg;
                
		Integer partialSum = (Integer) integerOI.getPrimitiveJavaObject(partial);
                
		LetterSumAgg myagg2 = new LetterSumAgg();
                
		myagg2.add(partialSum);
		myagg1.add(myagg2.sum);
	}
}

Terminate()函數返回AggregationBuffer中的內容，這裏產生了最終結果。

public Object terminate(AggregationBuffer agg) throws HiveException {
	LetterSumAgg myagg = (LetterSumAgg) agg;
	total = myagg.sum;
	return myagg.sum;
}

使用自定義函數

ADD JAR ./hive-extension-examples-master/target/hive-extensions-1.0-SNAPSHOT-jar-with-dependencies.jar;
CREATE TEMPORARY FUNCTION letters as ‘com.matthewrathbone.example.TotalNumOfLettersGenericUDAF‘;

SELECT letters(name) FROM people;
OK
44
Time taken: 20.688 seconds

資料參考

http://www.cnblogs.com/ggjucheng/archive/2013/02/01/2888051.html

http://blog.csdn.net/duguduchong/article/details/8684963
來源： http://lib.csdn.net/article/hive/42862

Hive UDAF開發詳解

相關推薦

Hive UDAF開發詳解

-s 聚合 而且 pri ros cal 關系 方法調用 evaluator 明這篇文章是來自Hadoop Hive UDAF Tutorial - Extending Hive with Aggregation Functions：的不嚴格翻譯，因為翻譯的文章示例寫得比較

Vue.js 插件開發詳解

js 前言隨著 Vue.js 越來越火，Vue.js 的相關插件也在不斷的被貢獻出來，數不勝數。比如官方推薦的 vue-router、vuex 等，都是非常優秀的插件。但是我們更多的人還只停留在使用的階段，比較少自己開發。所以接下來會通過一個簡單的 vue-toast 插件，來了解掌握插件的開發和使用。

斑布生活家系統開發詳解模式設計

log 所有 http 裏的 詳解 啟動 一次 什麽 氣體 斑布生活家系統開發(李想.185.6504.8478)根據化石研究，史前時代的始祖鳥被認為是最早的飛行鳥類，它在許多方面已顯現鳥的一些雛形，例如全身長有羽毛和翅膀、具有明顯的叉骨等。這些證據表明，鳥類的飛行能力應該

008-Hadoop Hive sql語法詳解3-DML 操作:元數據存儲

pan 查詢 寫入 所有 not insert語句 int 寫入文件 文件系統 一、概述 hive不支持用insert語句一條一條的進行插入操作，也不支持update操作。數據是以load的方式加載到建立好的表中。數據一旦導入就不可以修改。 DML包括：INSERT插入

果園農場種植遊戲復利拆分模式app系統開發詳解

計算 數量 政策 經營者 終端 默認 自己 產生 後臺 330復利拆分果園遊戲app系統開發（蘇公子.188.1414.7927）皮皮果遊戲開發，玫瑰莊園遊戲開發，復利拆分遊戲開發，330模式開發，330果園系統開發，復利拆分農場系統開發，英倫果開發，皮皮果遊戲介紹，地點的

火爆的車享家模式系統開發詳解

目前 聯合體 移動互聯網 體系 通過 系列 獲得 com img 車享家是一個關於用車的全方位平臺！上線至今都十分火熱，盈利能力也強,通過結合“高效線上運營”和“優質線下服務”，全方位打通汽車“看、選、買、用、賣”，簡單來看，是淘寶+o2o的聯合體。目前汽車市場蓬勃發展，汽

Android"掛逼"修煉之行---微信中定位照片的位置信息插件開發詳解

頁面 blank 如果 put 法律 mali 特殊字符 在哪裏 加載 一、前言 最近關於微信中，朋友之間發送原圖就可能暴露你的位置信息，其實這個問題不在於微信，微信是為了更好的體驗效果，才有發送原圖功能，而對於拍照，發送普通圖片微信後臺都會過濾圖片的exif信息，這樣就

android黑科技系列——微信定位聊天記錄中照片的位置信息插件開發詳解

poi 圖片查看 遍歷 什麽 term 反射 還需要 資料 避免 一、前言 最近關於微信中，朋友之間發送原圖就可能暴露你的位置信息，其實這個問題不在於微信，微信是為了更好的體驗效果，才有發送原圖功能，而對於拍照，發送普通圖片微信後臺都會過濾圖片的exif信息，這樣就不會攜

017-Hadoop Hive sql語法詳解7-去重排序、數據傾斜

col 去重排序 sel cluster 可能 更多 分發 指定 clust 一、數據去重排序 1.1、去重 　　distinct與group by 　　盡量避免使用distinct進行排重，特別是大表操作，用group by代替　　 -- 不建議 selec

016-Hadoop Hive sql語法詳解6-job輸入輸出優化、數據剪裁、減少job數、動態分區

分享 hive table 取數 nbsp put union 正在 style 一、job輸入輸出優化 善用muti-insert、union all，不同表的union all相當於multiple inputs，同一個表的union all，相當map一次輸出多條

PHP與Java集成開發詳解（一）

new 編程語言 到你 其中 web-inf request 測試 add 輸入 很久以前，有人從www上看到看到天空上一個很亮的亮點，它就是Java語言，與此同時，在另一個地方一位夢想家也看到了一個亮點，它就是PHP。 時間一天天過去，這兩個亮點也變得越來越亮，很快，它

【轉】Quartz.net持久化與集群部署開發詳解

疑惑 sum 常用 drive wid res net github hub 轉自：http://www.cnblogs.com/knowledgesea/p/5145239.html 序言 我前邊有幾篇文章有介紹過quartz的基本使用語法與類庫。但是他的執行計劃都是被寫

【轉】Android Camera 相機開發詳解

exc troy start 當前 container rac google getconf 對比度 在Android 5.0（SDK 21）中，Google使用Camera2替代了Camera接口。Camera2在接口和架構上做了巨大的變動, 但是基於眾所周知的原因

HTML5+Canvas開發詳解（第2版）.pdf

繪圖 blog ipa img script 必備 交互式 src 教程 通過HTML5+Canvas開發詳解（第2版），你將學到如何使用Canvas進行繪圖、渲染文字、處理圖像、創建動畫，而這些是開發交互式Web遊戲的必備知識。 本書針對Canvas和HTML5技

Java中的OA信用盤源碼開發詳解

直接 所有 stat 開始 println printf void 一個 [] 首先先創建一個打印數組的方法，OA信用盤源碼開發（企 娥:217 1793 408）方便後面直接使用 public static void output(int []a) {for(int i=

EasyPR--開發詳解（8）文字定位

dont bubuko 通用 設置 光照 detect improve nmp easy 轉自https://www.cnblogs.com/subconscious/p/5637735.html 今天我們來介紹車牌定位中的一種新方法--文字定位方法（MSER），包括其主要

大資料篇：Hive的安裝詳解

hive是什麼？ 由facebook開源，用於解決海量結構化日誌的資料統計； 基於hadoop的一個數據倉庫工具，使用HDFS進行儲存並將結構化資料檔案對映成一張表，並提供類sql查詢的功能，其底層採用MR進行計算； 本質是將HQL轉化成MR程式。

ZCU106開發詳解之VIVADO開發環境的安裝

感謝大家漫長的等待！！ 我們團隊從去年底拿到ZCU106後就一直在進行相關研發，由於手頭上的活比較多就把ZCU106開發詳解的釋出給延遲了。 從本週開始我們團隊開始對我們在ZCU106開發過程中遇到的問題和解決辦法跟大家進行分享。我們的QQ群是836535064。希望跟大家一起玩好這一個牛逼的晶片。

Spring Boot中使用MyBatis註解配置開發詳解

Spring Boot中使用MyBatis註解配置詳解 原創   2018-04-03 宗野       Spring Boot 最近專案原因可能會繼續開始使用

百惠優美模式APP開發詳解

百惠優美模式開發【張玲：⒈⒌⒍微⒉⒉⒎⒏電⒈⒌⒉⒏】，百惠優美模式商城系統開發，百惠優美系統平臺開發，百惠優美商城APP開發 Resource creation-資源創造。資源建立可能涉及分配新記憶體，將其對映到您的程序，並將其填充為零。通過從較大的堆或由堆支援的可迴圈資源記憶體建立資源，可以降低此成本。