hive GenericUDF1

时间：2023-03-09 07:03:43
hive GenericUDF1

和UDF相比，通用GDF（GenericUDF）支持复杂类型（比如List，struct等）的输入和输出。

下面来看一个小示例。

Hive中whereme表中包含若干人的行程如下：

  1. A       2013-10-10 8:00:00      home
  2. A       2013-10-10 10:00:00     Super Market
  3. A       2013-10-10 12:00:00     KFC
  4. A       2013-10-10 15:00:00     school
  5. A       2013-10-10 20:00:00     home
  6. A       2013-10-15 8:00:00      home
  7. A       2013-10-15 10:00:00     park
  8. A       2013-10-15 12:00:00     home
  9. A       2013-10-15 15:30:00     bank
  10. A       2013-10-15 19:00:00     home

通过查询我们要得到如下结果：

  1. A   2013-10-10  08:00:00    home    10:00:00    Super Market
  2. A   2013-10-10  10:00:00    Super Market    12:00:00    KFC
  3. A   2013-10-10  12:00:00    KFC 15:00:00    school
  4. A   2013-10-10  15:00:00    school  20:00:00    home
  5. A   2013-10-15  08:00:00    home    10:00:00    park
  6. A   2013-10-15  10:00:00    park    12:00:00    home
  7. A   2013-10-15  12:00:00    home    15:30:00    bank
  8. A   2013-10-15  15:30:00    bank    19:00:00    home

1.编写GenericUDF.

  1. package com.wz.udf;
  2. import org.apache.hadoop.io.Text;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.IntWritable;
  5. import org.apache.hadoop.io.FloatWritable;
  6. import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
  7. import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
  8. import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
  9. import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
  10. import org.apache.hadoop.hive.ql.metadata.HiveException;
  11. import org.apache.hadoop.hive.serde2.lazy.LazyString;
  12. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
  13. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
  14. import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;
  15. import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
  16. import org.apache.hadoop.hive.serde2.objectinspector.StandardListObjectInspector;
  17. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
  18. import org.apache.hadoop.hive.serde2.objectinspector.StructField;
  19. import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
  20. import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
  21. import org.apache.hadoop.hive.serde2.objectinspector.primitive.LongObjectInspector;
  22. import org.apache.hadoop.hive.serde2.objectinspector.primitive.IntObjectInspector;
  23. import org.apache.hadoop.hive.serde2.objectinspector.primitive.FloatObjectInspector;
  24. import org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector;
  25. import java.text.DateFormat;
  26. import java.text.SimpleDateFormat;
  27. import java.util.Date;
  28. import java.util.Calendar;
  29. import java.util.ArrayList;
  30. public class helloGenericUDF extends GenericUDF {
  31. ////输入变量定义
  32. private ObjectInspector peopleObj;
  33. private ObjectInspector timeObj;
  34. private ObjectInspector placeObj;
  35. //之前记录保存
  36. String strPreTime = "";
  37. String strPrePlace = "";
  38. String strPrePeople = "";
  39. @Override
  40. //1.确认输入类型是否正确
  41. //2.输出类型的定义
  42. public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
  43. peopleObj = (ObjectInspector)arguments[0];
  44. timeObj = (ObjectInspector)arguments[1];
  45. placeObj = (ObjectInspector)arguments[2];
  46. //输出结构体定义
  47. ArrayList structFieldNames = new ArrayList();
  48. ArrayList structFieldObjectInspectors = new ArrayList();
  49. structFieldNames.add("people");
  50. structFieldNames.add("day");
  51. structFieldNames.add("from_time");
  52. structFieldNames.add("from_place");
  53. structFieldNames.add("to_time");
  54. structFieldNames.add("to_place");
  55. structFieldObjectInspectors.add( PrimitiveObjectInspectorFactory.writableStringObjectInspector );
  56. structFieldObjectInspectors.add( PrimitiveObjectInspectorFactory.writableStringObjectInspector );
  57. structFieldObjectInspectors.add( PrimitiveObjectInspectorFactory.writableStringObjectInspector );
  58. structFieldObjectInspectors.add( PrimitiveObjectInspectorFactory.writableStringObjectInspector );
  59. structFieldObjectInspectors.add( PrimitiveObjectInspectorFactory.writableStringObjectInspector );
  60. structFieldObjectInspectors.add( PrimitiveObjectInspectorFactory.writableStringObjectInspector );
  61. StructObjectInspector si2;
  62. si2 = ObjectInspectorFactory.getStandardStructObjectInspector(structFieldNames, structFieldObjectInspectors);
  63. return si2;
  64. }
  65. //遍历每条记录
  66. @Override
  67. public Object evaluate(DeferredObject[] arguments) throws HiveException{
  68. LazyString LPeople = (LazyString)(arguments[0].get());
  69. String strPeople = ((StringObjectInspector)peopleObj).getPrimitiveJavaObject( LPeople );
  70. LazyString LTime = (LazyString)(arguments[1].get());
  71. String strTime = ((StringObjectInspector)timeObj).getPrimitiveJavaObject( LTime );
  72. LazyString LPlace = (LazyString)(arguments[2].get());
  73. String strPlace = ((StringObjectInspector)placeObj).getPrimitiveJavaObject( LPlace );
  74. Object[] e;
  75. e = new Object[6];
  76. try
  77. {
  78. //如果是同一个人，同一天
  79. if(strPrePeople.equals(strPeople) && IsSameDay(strTime) )
  80. {
  81. e[0] = new Text(strPeople);
  82. e[1] = new Text(GetYearMonthDay(strTime));
  83. e[2] = new Text(GetTime(strPreTime));
  84. e[3] = new Text(strPrePlace);
  85. e[4] = new Text(GetTime(strTime));
  86. e[5] = new Text(strPlace);
  87. }
  88. else
  89. {
  90. e[0] = new Text(strPeople);
  91. e[1] = new Text(GetYearMonthDay(strTime));
  92. e[2] = new Text("null");
  93. e[3] = new Text("null");
  94. e[4] = new Text(GetTime(strTime));
  95. e[5] = new Text(strPlace);
  96. }
  97. }
  98. catch(java.text.ParseException ex)
  99. {
  100. }
  101. strPrePeople = new String(strPeople);
  102. strPreTime= new String(strTime);
  103. strPrePlace = new String(strPlace);
  104. return e;
  105. }
  106. @Override
  107. public String getDisplayString(String[] children) {
  108. assert( children.length>0 );
  109. StringBuilder sb = new StringBuilder();
  110. sb.append("helloGenericUDF(");
  111. sb.append(children[0]);
  112. sb.append(")");
  113. return sb.toString();
  114. }
  115. //比较相邻两个时间段是否在同一天
  116. private boolean IsSameDay(String strTime) throws java.text.ParseException{
  117. if(strPreTime.isEmpty()){
  118. return false;
  119. }
  120. String curDay = GetYearMonthDay(strTime);
  121. String preDay = GetYearMonthDay(strPreTime);
  122. return curDay.equals(preDay);
  123. }
  124. //获取年月日
  125. private String GetYearMonthDay(String strTime)  throws java.text.ParseException{
  126. DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  127. Date curDate = df.parse(strTime);
  128. df = new SimpleDateFormat("yyyy-MM-dd");
  129. return df.format(curDate);
  130. }
  131. //获取时间
  132. private String GetTime(String strTime)  throws java.text.ParseException{
  133. DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  134. Date curDate = df.parse(strTime);
  135. df = new SimpleDateFormat("HH:mm:ss");
  136. return df.format(curDate);
  137. }
  138. }

2.在Hive里面创建两张表，一张包含结构体的表保存执行GenericUDF查询后的结果，另外一张用于保存最终结果.

  1. hive> create table whereresult(people string,day string,from_time string,from_place string,to_time string,to_place string);
  2. OK
  3. Time taken: 0.287 seconds
  4. hive> create table tmpResult(info struct<people:string,day:string,from_time:str>ing,from_place:string,to_time:string,to_place:string>);
  5. OK
  6. Time taken: 0.074 seconds

3.执行GenericUDF查询,得到最终结果。

    1. hive> insert overwrite table tmpResult select hellogenericudf(whereme.people,whereme.time,whereme.place) from whereme;
    2. hive> insert overwrite table whereresult select info.people,info.day,info.from_time,info.from_place,info.to_time,info.to_place from tmpResult where info.from_time<>'null';
    3. Total MapReduce jobs = 2
    4. Launching Job 1 out of 2
    5. Number of reduce tasks is set to 0 since there's no reduce operator
    6. Starting Job = job_201312022129_0006, Tracking URL = http://localhost:50030/jobdetails.jsp?jobid=job_201312022129_0006
    7. Kill Command = /home/wangzhun/hadoop/hadoop-0.20.2/bin/../bin/hadoop job  -Dmapred.job.tracker=localhost:9001 -kill job_201312022129_0006
    8. Hadoop job information for Stage-1: number of mappers: 1; number of reducers: 0
    9. 2013-12-02 22:48:40,733 Stage-1 map = 0%,  reduce = 0%
    10. 2013-12-02 22:48:49,825 Stage-1 map = 100%,  reduce = 0%
    11. 2013-12-02 22:48:52,869 Stage-1 map = 100%,  reduce = 100%
    12. Ended Job = job_201312022129_0006
    13. Ended Job = -383357832, job is filtered out (removed at runtime).
    14. Moving data to: hdfs://localhost:9000/tmp/hive-root/hive_2013-12-02_22-48-24_406_2701579121398466034/-ext-10000
    15. Loading data to table default.whereresult
    16. Deleted hdfs://localhost:9000/user/hive/warehouse/whereresult
    17. Table default.whereresult stats: [num_partitions: 0, num_files: 1, num_rows: 0, total_size: 346, raw_data_size: 0]
    18. 8 Rows loaded to whereresult
    19. MapReduce Jobs Launched:
    20. Job 0: Map: 1   HDFS Read: 420 HDFS Write: 346 SUCESS
    21. Total MapReduce CPU Time Spent: 0 msec
    22. OK
    23. Time taken: 29.098 seconds
    24. hive> select * from whereresult;
    25. OK
    26. A   2013-10-10  08:00:00    home    10:00:00    Super Market
    27. A   2013-10-10  10:00:00    Super Market    12:00:00    KFC
    28. A   2013-10-10  12:00:00    KFC 15:00:00    school
    29. A   2013-10-10  15:00:00    school  20:00:00    home
    30. A   2013-10-15  08:00:00    home    10:00:00    park
    31. A   2013-10-15  10:00:00    park    12:00:00    home
    32. A   2013-10-15  12:00:00    home    15:30:00    bank
    33. A   2013-10-15  15:30:00    bank    19:00:00    home
    34. Time taken: 0.105 seconds

相关文章