Hive自定义UDF和聚合函数UDAF

Hive是一种构建在Hadoop上的数据仓库,Hive把SQL查询转换为一系列在Hadoop集群中运行的MapReduce作业,是MapReduce更高层次的抽象,不用编写具体的MapReduce方法。Hive将数据组织为表,这就使得HDFS上的数据有了结构,元数据即表的模式,都存储在名为metastore的数据库中。
可以在hive的外壳环境中直接使用dfs访问hadoop的文件系统命令。
       Hive可以允许用户编写自己定义的函数UDF,来在查询中使用。Hive中有3种UDF:
       UDF:操作单个数据行,产生单个数据行;
       UDAF:操作多个数据行,产生一个数据行。
       UDTF:操作一个数据行,产生多个数据行一个表作为输出。
用户构建的UDF使用过程如下:
      第一步:继承UDF或者UDAF或者UDTF,实现特定的方法。
      第二步:将写好的类打包为jar。如hivefirst.jar.
      第三步:进入到Hive外壳环境中,利用add jar /home/hadoop/hivefirst.jar.注册该jar文件
      第四步:为该类起一个别名,create temporary function mylength as 'com.whut.StringLength';这里注意UDF只是为这个Hive会话临时定义的。
      第五步:在select中使用mylength(); 
自定义UDF
packagewhut;
importorg.apache.commons.lang.StringUtils;
importorg.apache.hadoop.hive.ql.exec.UDF;
importorg.apache.hadoop.io.Text;
//UDF是作用于单个数据行,产生一个数据行
//用户必须要继承UDF,且必须至少实现一个evalute方法,该方法并不在UDF中
//但是Hive会检查用户的UDF是否拥有一个evalute方法
publicclassStrip extendsUDF{
privateText result=newText();
//自定义方法
publicText evaluate(Text str)
{
if(str==null)
returnnull;
result.set(StringUtils.strip(str.toString()));
returnresult;
}
publicText evaluate(Text str,String stripChars)
{
if(str==null)
returnnull;
result.set(StringUtils.strip(str.toString(),stripChars));
returnresult;
}
}

注意事项:
   1,一个用户UDF必须继承org.apache.hadoop.hive.ql.exec.UDF;
   2,一个UDF必须要包含有evaluate()方法,但是该方法并不存在于UDF中。evaluate的参数个数以及类型都是用户自己定义的。在使用的时候,Hive会调用UDF的evaluate()方法。 
自定义UDAF
该UDAF主要是找到最大值
 
packagewhut;
importorg.apache.hadoop.hive.ql.exec.UDAF;
importorg.apache.hadoop.hive.ql.exec.UDAFEvaluator;
importorg.apache.hadoop.io.IntWritable;
//UDAF是输入多个数据行,产生一个数据行
//用户自定义的UDAF必须是继承了UDAF,且内部包含多个实现了exec的静态类
publicclassMaxiNumber extendsUDAF{
publicstaticclassMaxiNumberIntUDAFEvaluator implementsUDAFEvaluator{
//最终结果
privateIntWritable result;
//负责初始化计算函数并设置它的内部状态,result是存放最终结果的
@Override
publicvoidinit() {
result=null;
}
//每次对一个新值进行聚集计算都会调用iterate方法
publicbooleaniterate(IntWritable value)
{
if(value==null)
returnfalse;
if(result==null)
result=newIntWritable(value.get());
else
result.set(Math.max(result.get(), value.get()));
returntrue;
}

//Hive需要部分聚集结果的时候会调用该方法
//会返回一个封装了聚集计算当前状态的对象
publicIntWritable terminatePartial()
{
returnresult;
}
//合并两个部分聚集值会调用这个方法
publicbooleanmerge(IntWritable other)
{
returniterate(other);
}
//Hive需要最终聚集结果时候会调用该方法
publicIntWritable terminate()
{
returnresult;
}
}
}

注意事项:
    1,用户的UDAF必须继承了org.apache.hadoop.hive.ql.exec.UDAF;
    2,用户的UDAF必须包含至少一个实现了org.apache.hadoop.hive.ql.exec的静态类,诸如常见的实现了 UDAFEvaluator。
    3,一个计算函数必须实现的5个方法的具体含义如下:
    init():主要是负责初始化计算函数并且重设其内部状态,一般就是重设其内部字段。一般在静态类中定义一个内部字段来存放最终的结果。
   iterate():每一次对一个新值进行聚集计算时候都会调用该方法,计算函数会根据聚集计算结果更新内部状态。当输入值合法或者正确计算了,则就返回true。
   terminatePartial():Hive需要部分聚集结果的时候会调用该方法,必须要返回一个封装了聚集计算当前状态的对象。
   merge():Hive进行合并一个部分聚集和另一个部分聚集的时候会调用该方法。
   terminate():Hive最终聚集结果的时候就会调用该方法。计算函数需要把状态作为一个值返回给用户。

0 个评论

要回复文章请先登录注册