1. 前言
在实际的数据处理中，我们经常需要对数据进行各种各样的计算和处理，例如字符串的拼接、日期的转换、数值的运算等等。Hive作为一款基于Hadoop生态圈的数据仓库工具，提供了UDF（User-Defined Function）机制，使得用户可以通过编写自定义函数来满足不同的数据处理需求。本文将介绍如何利用Java和Scala编写Hive UDF函数，实现解析存储在Hive库中Json格式的字符串。
2. UDF函数逻辑简介
UDF函数主要解析存储在Hive中Json格式字符串的，通过参数控制输出Json串的所有的key值或者value值。如下： 输入：
--UDF函数参数
analyse_hive_json(jsonObj,resultType,splitType)
jsonObj String类型，json格式字符串
resultType Int类型，返回结果类型，0表示返回key串，1表示返回value串
splitType Int类型，返回结果分隔符，0表示以“，”分割，1表示以“ ”分割
--例如：
select analyse_hive_json ('{"kobe":"goat","rusell":"mvp","rose":"fast"}',0,0) ;
输出：
kobe,rusell,rose
3. 准备工作
在编写Hive UDF函数之前，需要准备好以下环境： JDK 1.8或以上版本 Scala 2.11或以上版本 Hive 1.2或以上版本
4. 编写Hive UDF函数
4.1 第一步：创建工程
首先，需要在本地或者集群上创建一个Scala和Java工程。可以使用Maven或者SBT等构建工具来管理依赖和构建项目。在创建工程的过程中，需要加入Hive和Hadoop相关的依赖，例如：
4.2 第二步：编写Hive UDF函数
在创建好工程后，可以开始编写Hive UDF函数。
4.2.1 Java代码展示
package com.zero.java;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hive.ql.exec.UDF;
import java.util.ArrayList;
import java.util.Iterator;
public class AnalyseHiveJson extends UDF {
/**
*
* @param jsonObj json格式字符串
* @param splitType 返回结果分隔符，0表示以“，”分割，1表示以“ ”分割
* @param resultType 返回结果类型，0表示返回key串，1表示返回value串
* @return
*/
public static String evaluate(Object jsonObj,Integer resultType ,Integer splitType ){
if (jsonObj == null) return null;
String jsonStr = String.valueOf(jsonObj).replace(" ","");
ArrayList keyList = new ArrayList
ArrayList valueList = new ArrayList
String result ="";
try{
//判断josnStr的长度是否小于7
if(jsonStr.length()<=7) return result; else {
//使用alibaba.fastjson将json格式字符串转成JsonObject
JSONObject asJsonObj = JSON.parseObject(jsonStr);
//遍历JsonObject;
Iterator itor = asJsonObj.keySet().iterator();
String key = "";
String value = "";
while (itor.hasNext()){
key = itor.next().toString();
value = asJsonObj.getString(key);
keyList.add(key);
valueList.add(value);
}
//返回结果处理：0-key串，1-value串
if(resultType == 0){
if(splitType == 0){
result = StringUtils.join(keyList, ",");
}else{
result = StringUtils.join(keyList, " ");
}
} else if(resultType == 1){
if(splitType == 0){
result = StringUtils.join(valueList, " ");
}else{
result = StringUtils.join(valueList, ",");
}
}
}
return result;
}catch (Exception e){
return null;
}
}
}
测试运行结果
public static void main(String[] args) {
String str = "{\"kobe\":\"goat\",\"rusell\":\"mvp\",\"rose\":\"fast\"}";
System.out.println(evaluate(str,0,0));
System.out.println(evaluate(str,0,1));
System.out.println(evaluate(str,1,0));
System.out.println(evaluate(str,1,1));
}
kobe,rusell,rose
kobe rusell rose
goat mvp fast
goat,mvp,fast
4.2.2 Scala代码展示
package com.zero.scala
import com.alibaba.fastjson.JSON
import org.apache.hadoop.hive.ql.exec.UDF
import scala.collection.mutable.ArrayBuffer
object AnalyseHiveJson extends UDF{
/**
*
* @param jsonObj json格式字符串
* @param splitType 返回结果分隔符，0表示以“，”分割，1表示以“ ”分割
* @param resultType 返回结果类型，0表示返回key串，1表示返回value串
* @return
*/
def evaluate(jsonObj : Object, resultType : Int, splitType : Int):String={
if (jsonObj == null) return null
val jsonStr = String.valueOf(jsonObj).replace(" ","")
val keyList = new ArrayBuffer[String]
val valueList = new ArrayBuffer[String]
var result =""
try{
//判断josnStr的长度是否小于7
if(jsonStr.length<=7) return result else {
//使用alibaba.fastjson将json格式字符串转成JsonObject
val asJsonObj = JSON.parseObject(jsonStr)
//遍历JsonObject
val itor = asJsonObj.keySet().iterator()
while (itor.hasNext){
val in = itor.next()
keyList += in
valueList += asJsonObj.getString(in)
}
//返回结果处理：0-key串，1-value串
if(resultType == 0){
if(splitType == 0){
result = keyList.mkString(" ")
}else{
result = keyList.mkString(",")
}
} else if(resultType == 1){
if(splitType == 0){
result = valueList.mkString(" ")
}else{
result = valueList.mkString(",")
}
}
}
result
}finally {
null
}
}
}
测试运行结果
def main(args: Array[String]): Unit = {
val str = "{\"kobe\":\"goat\",\"rusell\":\"mvp\",\"rose\":\"fast\"}"
println(evaluate(str,0,0))
println(evaluate(str,0,1))
println(evaluate(str,1,0))
println(evaluate(str,1,1))
}
kobe,rusell,rose
kobe rusell rose
goat mvp fast
goat,mvp,fast
4.3 第三步：打包工程
在编写好Hive UDF函数后，需要将工程打包成jar文件，以便在Hive中使用。使用maven直接打包，或者直接在Idea中Build Artifacts，可以在工程的target目录下找到打包好的jar文件。
4.4 第四步：上传jar文件到Hive
在打包后，需要将jar文件上传到Hive的classpath或者HDFS中，以便在Hive SQL语句中使用自定义函数。可以使用以下命令将jar文件上传到HDFS中：
hadoop fs -put -f analyse-hive-json.jar /user/hdfs/zero/libs/
4.5 第五步：使用Hive UDF函数
在完成前面的准备工作后，就可以在Hive SQL语句中使用自定义函数了：
--添加jar包至环境变量
add jar /user/hdfs/zero/libs/analyse-hive-json.jar;
--创建自定义函数
create function analyse_hive_json as 'com.zero.java.AnalyseHiveJson';
--使用函数
select analyse_hive_json ('{"kobe":"goat","rusell":"mvp","rose":"fast"}',0,0) ;
select analyse_hive_json ('{"kobe":"goat","rusell":"mvp","rose":"fast"}',0,1) ;
select analyse_hive_json ('{"kobe":"goat","rusell":"mvp","rose":"fast"}',1,0) ;
select analyse_hive_json ('{"kobe":"goat","rusell":"mvp","rose":"fast"}',1,1) ;
结果展示
kobe,rusell,rose
kobe rusell rose
goat mvp fast
goat,mvp,fast
5 总结
本文介绍了如何使用Java、Scala编写Hive UDF函数。需要注意的是，在使用自定义函数之前，需要先将工程打包成jar文件，并将jar文件上传到Hive的classpath或者HDFS中。然后就可以在Hive SQL语句中使用自定义函数了。Hive UDF函数的编写可以根据具体的需求进行，可以实现不同的数据处理逻辑。
好文阅读
发表评论