玩轉大數據系列之Apache Pig如何經過自定義UDF查詢數據庫（五）

GMV(必定時間內的成交總額)是一個衡量電商網站營業收入的一項重要指標，例如淘寶，京東都有這樣的衡量標準，感興趣的朋友能夠本身科普下這方面的概念知識。

固然散仙今天，並非來解釋概念的，而是記錄下最近工做的一些東西，原來咱們平臺的GMV只有一個總的成交金額，並無細分到各個系統的GMV的比重，好比搜索端，推薦端，移動端等等。

經過細粒度的分析各個系統所佔的比重，對於指導各個系統完善和發展有必定的重要意義，這裏不就深說了，下面先來看下散仙分析的搜索gmv的數據佈局方式。


（1）Hadoop集羣上，存儲了一些非核心的數據，好比訪問數據，點擊數據，購物車數據，下單數據（這個是從數據庫裏天天同步到HDFS上的，算是備份吧）
（2）Oracle數據庫中，存儲了訂單信息，交易信息，商品信息，支付信息等一些電商的核心數據

其實關於gmv的計算方式，在咱們oracle庫裏，以及有一個存儲過程封裝了複雜的細節的處理，包括運費，折扣，不一樣國家，不一樣地域，信用用戶，等等，在使用時候，只須要傳入一個訂單編號便可，計算出本單的gmv成交金額。


這樣以來的，按照目前的數據狀況，訂單編號是從Hadoop集羣上，一直是從搜索，點擊，添加購物車，下單計算出來的，而後獲取的對應的訂單編號，注意這個過程當中，是須要全程去爬蟲數據的，由於還要算最終的GMV成交額，因此須要找到必定時期內的訂單號，而後經過調用在oracle庫的封裝好的函數，計算出gmv，這樣以來，就可以比較細跟蹤各個階段運行軌跡和成交額。

ok，業務上的分析大體如此，下面就看下，技術上如何實現，其實就是須要Pig的一個自定義UDF函數，在遍歷每一行的recoder時，去查詢oracle只讀庫，獲取gmv的值，並將最終結果存儲起來，以圖形化方式展現。

Pig裏面對UDF函數很是豐富，比較經常使用的是轉化函數和加載存儲函數，這一點在Hive裏，也是如此，以前的文章中，散仙介紹過，經過自定義UDF將pig分析的結果直接存儲到數據庫或索引中，便於檢索和發揮不一樣框架之間的組合優點。

核心代碼以下：

Java代碼  

package com.pig.dhgate.getgvmbyrfxno;

import java.io.IOException;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * 自定義Pig UDF實現查詢db計算gmv
 * **/
public class GetGmvByRfxno extends EvalFunc<Double> {
	/**日誌對象*/
	static Logger log =LoggerFactory.getLogger(GetGmvByRfxno.class);
	/**數據庫工具類*/
	DBTools dbtools=new DBTools();
	
	@Override
	public Double exec(Tuple input) throws IOException {
		
		if(input!=null&&input.size()!=0){
			//獲取傳入的訂單號
			String rfxno =(String)input.get(0);
			//經過db類，查詢對應的gmv並返回
			double gmv=dbtools.getGmvByRfxno(rfxno);
			return gmv;
		}else{
			//對null，空值，一概按0處理
			return 0.00;
		}
	}
}

數據庫封裝類：

Java代碼  

/***
 * 數據庫工具類
 * */
public class DBTools {
	
	/**日誌對象*/
	static Logger log =LoggerFactory.getLogger(DBTools.class);
	
	
	private  static  Connection conn;
	private  static PreparedStatement ps;
	private   ResultSet rs;
	//從虛擬表查詢函數
	private static  String  sql="select datasql.GETGMV(?) as gmv  from dual ";
	static{
		try{
		Class.forName("oracle.jdbc.driver.OracleDriver");
		conn = DriverManager.getConnection("jdbc:oracle:thin:@ip地址:1521:數據庫名", "用戶名", "密碼");
		System.out.println("數據庫鏈接："+conn);
		ps=conn.prepareStatement(sql);
		}catch(Exception e){
			log.error("初始化oracle驅動異常!", e);
		}
	}
	
	/**根據一個rfxno獲取對應的產品的gmv
	 * **/
	public double getGmvByRfxno(String rfxno){
		try{
		ps.setString(1, rfxno);
		rs = ps.executeQuery();
		if(rs.next()){
			double gmv=rs.getDouble("gmv");
//			System.out.println("gmv是：  "+gmv);
			return gmv;
		}
		rs.close();
		}catch(Exception e){
			log.error("根據rfxno獲取gmv出錯!",e);
		}
		return 0.0;
	}
	}

其實，代碼仍是比較簡單的，在這裏，你能夠從任何數據源獲取須要的數據，而不單單是數據庫，你也能夠從redis，memcache，文件，xml，等等裏獲取須要組合用的數據。

遇到一個異常：在sql語句後面，不用加分號，相似下面的這樣的語句，經過jdbc編譯而後調用oracle是不經過的：

Sql代碼  

1. select datasql.GETGMV(?) as gmv  from dual;  

select datasql.GETGMV(?) as gmv  from dual;

這一點須要注意下。

最後來看下以下在pig腳本里，使用自定義的函數：
（1）使用ant打包自定義的udf函數的jar
（2）在pig腳本里，註冊相關的jar包，注意若是有依賴關係，依賴的jar包，也須要註冊，例如本例中的oracle的jdbc的驅動包
（3）在對應的地方，經過類的全路徑名，引用此函數，完成對應的查詢轉換，並將新獲得的一個字段，做爲原始一行記錄的字段擴充。

腳本以下：

Java代碼  

--註冊依賴的jar包
register /home/search/dongliang/nsconvent/checklist/ojdbc.jar
register /home/search/dongliang/nsconvent/checklist/tools.jar


--加載原有數據
m = load '/tmp/mdm/VW_TD_RFX' using PigStorage('\\x07');
--加載原有數據
n = load '/tmp/mdm/TD_RFX_PRODUCT' using PigStorage('\\x07');

--過濾出符合時間的數據

m= filter m by ToMilliSeconds(ToDate($3,'yyyy-MM-dd HH:mm:ss')) >= ToMilliSeconds(ToDate('$day 00:00:00','yyyy-MM-dd HH:mm:ss')) and ToMilliSeconds(ToDate($3
,'yyyy-MM-dd HH:mm:ss')) <= ToMilliSeconds(ToDate('$day 23:59:59','yyyy-MM-dd HH:mm:ss'))  ;

--提取相關字段，並完成計算
m = foreach m generate $0 as arfid, $1 as rfxno , com.pig.dhgate.getgvmbyrfxno.GetGmvByRfxno((chararray)$1) as gmv  , $4 as bid ;
--獲取topN數據
m = limit m 10 ;
--打印輸出
dump m;