首页 > 代码库 > 日活跃用户统计函数
日活跃用户统计函数
题记：
在做运营统计的时候，一个最常见的指标是日活跃用户数（DAU），它的一般性概念为当日所有用户的去重，但是在大部分情况下，我们获取到的数据中会有登录用户与有匿名用户，而这部分用户是会出现重叠的。常规的做法是利用cookie或者imei（移动端）进行自关联，然后算出有多少用户同时是登录用户和匿名用户，最终的 日活跃用户数 = 登录用户+匿名用户-匿名转登录用户。
在实际操作中需要写复杂的HQL才能完成这部分工作，而且运行效率低下，为此需要开发一个UDAF函数进行处理。
首先说明一下函数的原理：
/** * 根据flag,uid和imei信息计算个数 * -fla为1 ： 将对应的UID存储在UID集合中，该集合代表登录用户 * -flag不为1 ： 将对应的imei|wyy存储在IMEI集合中，该集合代表匿名用户 * 将imei|wyy存储一个Map当中，并且判断该imei|wyy对应的flag是否同时出现过0和1俩个值，如果是则map中对应的value = http://www.mamicode.com/2否则为flag>*/
其中flag参数可以用其它udf函数进行替换，用以判断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; // 存储中间结果 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); } }}
又有工作要忙了，先把代码放上来，下次再写分析
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。