hive: 自定义UDTF

比眉伴天荒 2022-04-22 08:46 263阅读 0赞

在hive项目中， 有client和server通信的log日志体系如下，中间的网络传输使用的json格式，所以在server端接收时需要使用阿里的fastJSON来解析为日志聚合体LogAgg类；然后使用自定义表生成函数UDTF, 分别取出Error表, Event表，StartUp表的数据, 数据转储到各表中，完成日志的初步ETL。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]  
所以，这个UDTF如何定义？  
类似于上篇的[自定义UDF函数][UDF]， 这里也需要继承一个hive类 org.apache.hadoop.hive.ql.udf.generic.GenericUDTF, 并实现三个方法即可（该类的描述如下：为单个输入行生成可变数量的输出行， 类似于explode(array)…）

# 自定义UDTF：函数的结构 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 2]  
初始化判断： initialize方法

<table> 
 <thead> 
  <tr> 
   <th>类型判断</th> 
   <th>ois[0].getCategory() != ObjectInspector.Category.PRIMITIVE</th> 
   <th>(((PrimitiveObjectInspector) ois[0])).getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.INT、STRING、LONG</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>参数–(映射)–转换器</td> 
   <td>准备转换器数组： converters = new ObjectInspectorConverters.Converter[4]</td> 
   <td>ObjectInspectorConverters.getConverter(ois[0], PrimitiveObjectInspectorFactory.javaIntObjectInspector、 javaLongObjectInspector、 javaStringObjectInspector );</td> 
  </tr> 
 </tbody> 
</table>

在hive中被调用的方法：process， 解析参数生成新的数据

<table> 
 <thead> 
  <tr> 
   <th>解析参数</th> 
   <th>返回结果</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>String servertimestr = (String) converters[0].convert(args[0])</td> 
   <td>forward(arr)</td> 
  </tr> 
 </tbody> 
</table>

# 自定义UDTF：伪代码 #

public abstract class BaseForkUDTF<T> extends GenericUDTF {  //initialize(),  process(),  close()
    	//属性	
    	private Class<T> clazz;
    	private ObjectInspectorConverters.Converter[] converters ;
    	List<String> fieldNames =   new ArrayList<String>() ; //字段名称列表	
    	List<ObjectInspector> ois = new ArrayList<ObjectInspector>()  ;//检查器列表	
    
    	//构造
    	//通过构造函数抽取子类的泛型化超类部分
    	public BaseForkUDTF(){
    		ParameterizedType type = (ParameterizedType) this.getClass().getGenericSuperclass();
    		clazz = (Class) type.getActualTypeArguments()[0];
    	}
    
    	//重写方法1
     	StructObjectInspector initialize(ObjectInspector[] args){
    		//判断: 参数类型
    	  	 //参数 <--->参数转换器
    		BeanInfo bi = Introspector.getBeanInfo(clazz) ;
    		PropertyDescriptor[] pps = bi.getPropertyDescriptors();//得到所有属性
    
    		for(PropertyDescriptor pp :pps){
    			//获取属性名,  get !=null, set !=null 											if(type == Long.class || type == long.class){
    					fieldNames.add(name) ;
    					ois.add(PrimitiveObjectInspectorFactory.javaLongObjectInspector) ;
    				}
    				else if(type == int.class || type ==Integer.class){
    					fieldNames.add(name);
    					ois.add(PrimitiveObjectInspectorFactory.javaIntObjectInspector);
    				}
    				else if(type == String.class){
    					fieldNames.add(name);											ois.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
    				}
    		}
    		return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames , ois) ;	
    	}
    	
    	//重写方法2：		
    	void process(Object[] args) {
    		//用转换器：取出参数的值	
    		String servertimestr = (String) converters[0].convert(args[0]);
    		Long clienttimems = (Long) converters[1].convert(args[1]);
    		//取出String clientip ,  String json
    		
    		//日志分析
    		json = json.replace("\\\"" , "\"") ;
    		AppLogAggEntity agg = JSONObject.parseObject(json , AppLogAggEntity.class) ;//消息传递： obj-->json-->obj
    		alignTime(agg , servertimestr , clienttimems) ;// 对齐： client时间（时区：    服务器时间utc-----client时间cst）
    		//===================定义回调函数：  空实现=================================
    		extraProcess(clientip) ;//解析ip:  ==>china, 山东, A市
    
    		//.=================== 日志叉分--->还原为error, event, startup.....===================
    		//定义abstract函数:  子类有各自的实现----（error:  从agg中取出错误日志...）
    		List<T> logs = getLogs(agg) ;
    		for(Object log : logs){
    			Object[] arr = new Object[fieldNames.size()] ;
    			int i = 0 ;
    
    			//此对象有几个字段====>取出： 对应的属性值
    			for(String fname : fieldNames){
    				PropertyDescriptor pp = new PropertyDescriptor(fname , clazz) ;
    				Method get = pp.getReadMethod() ;
    				if(get != null){
    					Object retValue = get.invoke(log) ;
    					arr[i] = retValue ;
    				}
    				i ++ ;
    			}
    			//转发对象，就是输出一行
    			forward(arr);
    		}		
    	}//end process()
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]: /images/20220422/0e3a449fd1984cd6835d15b37afa5863.png
[UDF]: https://blog.csdn.net/eyeofeagle/article/details/83688719
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]: /images/20220422/8cd027fe5dad4defb83e3f2a6fae0652.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 2]: /images/20220422/9f57f07cfac34878bdbad04e9eb1cc28.png