題記:
在做運(yùn)營(yíng)統(tǒng)計(jì)的時(shí)候,一個(gè)最常見(jiàn)的指標(biāo)是日活躍用戶數(shù)(DAU),它的一般性概念為當(dāng)日所有用戶的去重,但是在大部分情況下,我們獲取到的數(shù)據(jù)中會(huì)有登錄用戶與有匿名用戶,而這部分用戶是會(huì)出現(xiàn)重疊的。常規(guī)的做法是利用cookie或者imei(移動(dòng)端)進(jìn)行自關(guān)聯(lián),然后算出有多少用戶同時(shí)是登錄用戶和匿名用戶,最終的 日活躍用戶數(shù) = 登錄用戶+匿名用戶-匿名轉(zhuǎn)登錄用戶。
在實(shí)際操作中需要寫復(fù)雜的HQL才能完成這部分工作,而且運(yùn)行效率低下,為此需要開(kāi)發(fā)一個(gè)UDAF函數(shù)進(jìn)行處理。
首先說(shuō)明一下函數(shù)的原理:
/* * * 根據(jù)flag,uid和imei信息計(jì)算個(gè)數(shù) * -fla為1 : 將對(duì)應(yīng)的UID存儲(chǔ)在UID集合中,該集合代表登錄用戶 * -flag不為1 : 將對(duì)應(yīng)的imei|wyy存儲(chǔ)在IMEI集合中,該集合代表匿名用戶 * 將imei|wyy存儲(chǔ)一個(gè)Map當(dāng)中,并且判斷該imei|wyy對(duì)應(yīng)的flag是否同時(shí)出現(xiàn)過(guò)0和1倆個(gè)值,如果是則map中對(duì)應(yīng)的value = 2否則為flag * 參數(shù)原型: * int itemcount(flag,uid,imei) * 參數(shù)說(shuō)明: * flag: 1或者不為1 * uid: 用戶id * imei: 用戶的第二個(gè)參照標(biāo)識(shí)(imei|wyy|cookie) * * 返回值: * int類型,dau值 * * 使用示例: * > SELECT flag, uid, imei FROM test; * 1 uid1 imei1 * 1 uid2 imei1 * 0 uid3 imei3 * * > SELECT daucount(flag,uid,imei) FROM test; * 1 */
其中flag參數(shù)可以用其它udf函數(shù)進(jìn)行替換,用以判斷uid是否是登錄用戶。
下面是具體的代碼塊:
package yy.juefan.udaf; import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; import java.util.Map.Entry; import org.apache.hadoop.hive.ql.exec.Description; import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException; import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException; import org.apache.hadoop.hive.ql.metadata.HiveException; import org.apache.hadoop.hive.ql.parse.SemanticException; import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver; import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator; import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory; import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory; import org.apache.hadoop.hive.serde2.objectinspector.StandardListObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.StandardMapObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.StructField; import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.primitive.IntObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory; import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils; import org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector; import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo; import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; @Description ( name = "dau_count" , value = "_FUNC_(flag,uid,imei)" ) public class GenericDauCount extends AbstractGenericUDAFResolver { private static final boolean DEBUG = false ; private static final boolean TRACE = false ; @Override public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters) throws SemanticException { if (parameters.length != 3 ) { throw new UDFArgumentLengthException( "Exactly 3 argument is expected." ); } if (((PrimitiveTypeInfo) parameters[0]).getPrimitiveCategory() != PrimitiveCategory.INT) { throw new UDFArgumentTypeException(0 , "Only int argument is accepted, but " + parameters[0].getTypeName() + " is passed" ); } if (((PrimitiveTypeInfo) parameters[1]).getPrimitiveCategory() != PrimitiveCategory.STRING) { throw new UDFArgumentTypeException(1 , "Only string argument is accepted, but " + parameters[1].getTypeName() + " is passed" ); } if (((PrimitiveTypeInfo) parameters[2]).getPrimitiveCategory() != PrimitiveCategory.STRING) { throw new UDFArgumentTypeException(2 , "Only string argument is accepted, but " + parameters[2].getTypeName() + " is passed" ); } return new GenericDauCountEvaluator(); } public static class GenericDauCountEvaluator extends GenericUDAFEvaluator { // 封裝接口 StructField uidSetField; StructField imeiSetField; StructField imeiMapField; StructObjectInspector map2red; // for PARTIAL1 and COMPLETE IntObjectInspector flagIO; StringObjectInspector uidIO; StringObjectInspector imeiIO; // for PARTIAL2 and FINAL StandardListObjectInspector uidSetIO; StandardListObjectInspector imeiSetIO; StandardMapObjectInspector imeiMapIO;
private static class DivideAB implements AggregationBuffer { Set <String> uidSet; Set <String> imeiSet; Map <String, Integer> imeiMap; } @Override public AggregationBuffer getNewAggregationBuffer() throws HiveException { DivideAB dab = new DivideAB(); reset(dab); return dab; } @Override public void reset(AggregationBuffer agg) throws HiveException { DivideAB dab = (DivideAB) agg; dab.uidSet = new HashSet<String> (); dab.imeiSet = new HashSet<String> (); dab.imeiMap = new HashMap<String, Integer> (); } boolean warned = false ; @Override public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException { super .init(m, parameters); // input if (m == Mode.PARTIAL1 || m == Mode.COMPLETE) { // for iterate assert (parameters.length == 3 ); flagIO = (IntObjectInspector) parameters[0 ]; uidIO = (StringObjectInspector) parameters[1 ]; imeiIO = (StringObjectInspector) parameters[2 ]; } else { // for merge map2red = (StructObjectInspector) parameters[0 ]; uidSetField = map2red.getStructFieldRef("uidSet" ); imeiSetField = map2red.getStructFieldRef("imeiSet" ); imeiMapField = map2red.getStructFieldRef("imeiMap" ); uidSetIO = (StandardListObjectInspector) uidSetField .getFieldObjectInspector(); imeiSetIO = (StandardListObjectInspector) imeiSetField .getFieldObjectInspector(); imeiMapIO = (StandardMapObjectInspector) imeiMapField .getFieldObjectInspector(); } if (m == Mode.PARTIAL1 || m == Mode.PARTIAL2) { ArrayList <ObjectInspector> foi = new ArrayList<ObjectInspector> (); ArrayList <String> fname = new ArrayList<String> (); foi.add(ObjectInspectorFactory .getStandardListObjectInspector(PrimitiveObjectInspectorFactory.javaStringObjectInspector)); foi.add(ObjectInspectorFactory .getStandardListObjectInspector(PrimitiveObjectInspectorFactory.javaStringObjectInspector)); foi.add(ObjectInspectorFactory .getStandardMapObjectInspector( PrimitiveObjectInspectorFactory.javaStringObjectInspector, PrimitiveObjectInspectorFactory.javaIntObjectInspector)); fname.add( "uidSet" ); fname.add( "imeiSet" ); fname.add( "imeiMap" ); return ObjectInspectorFactory.getStandardStructObjectInspector( fname, foi); } else { return PrimitiveObjectInspectorFactory.javaLongObjectInspector; } } @Override public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException { if (parameters.length != 3 ) { return ; } DivideAB dab = (DivideAB) agg; int check = PrimitiveObjectInspectorUtils.getInt(parameters[0 ], flagIO); String uid = PrimitiveObjectInspectorUtils.getString(parameters[1 ], uidIO); String imei = PrimitiveObjectInspectorUtils.getString( parameters[ 2 ], imeiIO); if (check == 1) { // 登錄用戶 dab.uidSet.add(uid); } else { // 匿名用戶 dab.imeiSet.add(imei); } if (dab.imeiMap.containsKey(imei)) { int flag = dab.imeiMap.get(imei); if (flag < 2 && flag != check) { dab.imeiMap.put(imei, 2 ); } } else { dab.imeiMap.put(imei, check); } } @Override public Object terminatePartial(AggregationBuffer agg) throws HiveException { DivideAB myagg = (DivideAB) agg; // 存儲(chǔ)中間結(jié)果 Object[] partialResult = new Object[3 ]; partialResult[ 0] = new ArrayList<String> (myagg.uidSet); partialResult[ 1] = new ArrayList<String> (myagg.imeiSet); partialResult[ 2] = new HashMap<String, Integer> (myagg.imeiMap); return partialResult; } @SuppressWarnings( "unchecked" ) @Override public void merge(AggregationBuffer agg, Object partial) throws HiveException { if (partial != null ) { DivideAB dab = (DivideAB) agg; Object uidSet = map2red .getStructFieldData(partial, uidSetField); Object imeiSet = map2red.getStructFieldData(partial, imeiSetField); Object imeiMap = map2red.getStructFieldData(partial, imeiMapField); List <Object> uidlist = (List<Object> ) uidSetIO.getList(uidSet); System.err.println( "uidList = " + uidlist.size()); if (uidlist != null ) { System.err.println( "uidSet = " + dab.uidSet.size()); for (Object obj : uidlist) { dab.uidSet.add(obj.toString()); } } List <Object> imeilist = (List<Object> ) uidSetIO .getList(imeiSet); if (imeilist != null ) { for (Object obj : imeilist) { dab.imeiSet.add(obj.toString()); } } Map <String, Integer> imeimap = (Map<String, Integer> ) imeiMapIO .getMap(imeiMap); for (Entry<?, ?> ele : imeimap.entrySet()) { Object kobj = ele.getKey(); String key = kobj.toString(); Object vobj = ele.getValue(); Object val = vobj.toString(); if (dab.imeiMap.containsKey(key)) { int flag = dab.imeiMap.get(key); if (flag < 2 && flag != Integer.parseInt(val.toString())) { dab.imeiMap.put(key, 2 ); } } else { dab.imeiMap.put(key, Integer.parseInt(val.toString())); } } } } @Override public Object terminate(AggregationBuffer agg) throws HiveException { DivideAB dab = (DivideAB) agg; int mix = 0 ; for ( int val : dab.imeiMap.values()) { if (val == 2 ) { mix ++ ; } } return ( long ) (dab.uidSet.size() + dab.imeiSet.size() - mix); } } }
?
又有工作要忙了,先把代碼放上來(lái),下次再寫分析
更多文章、技術(shù)交流、商務(wù)合作、聯(lián)系博主
微信掃碼或搜索:z360901061

微信掃一掃加我為好友
QQ號(hào)聯(lián)系: 360901061
您的支持是博主寫作最大的動(dòng)力,如果您喜歡我的文章,感覺(jué)我的文章對(duì)您有幫助,請(qǐng)用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧,狠狠點(diǎn)擊下面給點(diǎn)支持吧,站長(zhǎng)非常感激您!手機(jī)微信長(zhǎng)按不能支付解決辦法:請(qǐng)將微信支付二維碼保存到相冊(cè),切換到微信,然后點(diǎn)擊微信右上角掃一掃功能,選擇支付二維碼完成支付。
【本文對(duì)您有幫助就好】元
