玩转大数据系列之Apache Pig如何通过自定义UDF查询数据库（五）

电玩女神 2022-08-07 04:55 148阅读 0赞

![点击查看原始大小图片][e8d55c72-f37b-388e-b3b1-9b2e0c0eb326.jpg]  
  
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代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif] ![收藏代码][icon_star.png]![spinner.gif][]

1.  package com.pig.dhgate.getgvmbyrfxno;
2.  
3.  import java.io.IOException;
4.  
5.  import org.apache.pig.EvalFunc;
6.  import org.apache.pig.data.Tuple;
7.  import org.slf4j.Logger;
8.  import org.slf4j.LoggerFactory;
9.  /\*\* 
10.  \* 自定义Pig UDF实现查询db计算gmv 
11.  \* \*\*/
12. public class GetGmvByRfxno extends EvalFunc<Double> \{
13. /\*\*日志对象\*/
14. static Logger log =LoggerFactory.getLogger(GetGmvByRfxno.class);
15. /\*\*数据库工具类\*/
16. DBTools dbtools=new DBTools();
17. 
18. @Override
19. public Double exec(Tuple input) throws IOException \{
20. 
21. if(input!=null&&input.size()!=0)\{
22. //获取传入的订单号 
23. String rfxno =(String)input.get(0);
24. //通过db类，查询对应的gmv并返回 
25. double gmv=dbtools.getGmvByRfxno(rfxno);
26. return gmv;
27. \}else\{
28. //对null，空值，一律按0处理 
29. return 0.00;
30. \}
31. \}
32. \}

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代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif] ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*\* 
2.   \* 数据库工具类 
3.   \* \*/
4.  public class DBTools \{
5.  
6.  /\*\*日志对象\*/
7.  static Logger log =LoggerFactory.getLogger(DBTools.class);
8.  
9.  
10. private static Connection conn;
11. private static PreparedStatement ps;
12. private ResultSet rs;
13. //从虚拟表查询函数 
14. private static String sql="select datasql.GETGMV(?) as gmv from dual ";
15. static\{
16. try\{
17. Class.forName("oracle.jdbc.driver.OracleDriver");
18. conn = DriverManager.getConnection("jdbc:oracle:thin:@ip地址:1521:数据库名", "用户名", "密码");
19. System.out.println("数据库连接："\+conn);
20. ps=conn.prepareStatement(sql);
21. \}catch(Exception e)\{
22. log.error("初始化oracle驱动异常!", e);
23. \}
24. \}
25. 
26. /\*\*根据一个rfxno获取对应的产品的gmv 
27.  \* \*\*/
28. public double getGmvByRfxno(String rfxno)\{
29. try\{
30. ps.setString(1, rfxno);
31. rs = ps.executeQuery();
32. if(rs.next())\{
33. double gmv=rs.getDouble("gmv");
34. // System.out.println("gmv是： "+gmv); 
35. return gmv;
36. \}
37. rs.close();
38. \}catch(Exception e)\{
39. log.error("根据rfxno获取gmv出错!",e);
40. \}
41. return 0.0;
42. \}
43. \}

/***
     * 数据库工具类
     * */
    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代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif] ![收藏代码][icon_star.png]![spinner.gif][]

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代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif] ![收藏代码][icon_star.png]![spinner.gif][]

1.  \--注册依赖的jar包
2.  register /home/search/dongliang/nsconvent/checklist/ojdbc.jar
3.  register /home/search/dongliang/nsconvent/checklist/tools.jar
4.  
5.  
6.  \--加载原有数据
7.  m = load '/tmp/mdm/VW\_TD\_RFX' using PigStorage('\\\\x07');
8.  \--加载原有数据
9.  n = load '/tmp/mdm/TD\_RFX\_PRODUCT' using PigStorage('\\\\x07');
10. 
11. \--过滤出符合时间的数据
12. 
13. 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
14. ,'yyyy-MM-dd HH:mm:ss')) <= ToMilliSeconds(ToDate('$day 23:59:59','yyyy-MM-dd HH:mm:ss')) ;
15. 
16. \--提取相关字段，并完成计算
17. m = foreach m generate $0 as arfid, $1 as rfxno , com.pig.dhgate.getgvmbyrfxno.GetGmvByRfxno((chararray)$1) as gmv , $4 as bid ;
18. \--获取topN数据
19. m = limit m 10 ;
20. \--打印输出
21. dump m;

--注册依赖的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;

[e8d55c72-f37b-388e-b3b1-9b2e0c0eb326.jpg]: /images/20220805/c97a2e355e7c4a01a58d153bbb0de223.png
[icon_copy.gif]: http://qindongliang.iteye.com/images/icon_copy.gif
[_icon_copy.gif]: http://qindongliang.iteye.com/blog/2191889#
[icon_star.png]: http://qindongliang.iteye.com/images/icon_star.png
[spinner.gif]: /images/20220805/b3df66e403ca4a36a017fb4605758a81.png