如何用Java编写继承AbstractGenericUDAFResolver的Hive UDAF函数parsed_user_agent

云原生阿岚

2026-5-15

没问题，我来帮你把这个Hive UDAF的实现逻辑理清楚——你已经掌握了User-Agent的解析方法，那核心就是搞定UDAF的输入输出定义、聚合流程这几块了。下面一步步来：

首先明确输入输出的类型定义

你的需求是按city分组，对每组的User_Agent字符串进行解析，最终返回包含操作系统、浏览器、设备信息的聚合结果。这里我们可以把输出定义为一个Hive StructType，里面包含三个Map（用来统计每组中各类型的出现次数，比如每个操作系统被多少用户使用）：

输入：单个String类型的User_Agent字段
输出：struct<os_stats:map<string,int>, browser_stats:map<string,int>, device_stats:map<string,int>>

第一步：实现继承AbstractGenericUDAFResolver的Resolver类

这个类的作用是向Hive声明UDAF支持的输入输出类型，核心是重写getEvaluator方法，返回我们自定义的聚合逻辑实现类。

import org.apache.hadoop.hive.ql.exec.UDAFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
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.*;
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.StringObjectInspector;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

public class ParsedUserAgentUDAF extends AbstractGenericUDAFResolver {

    @Override
    public GenericUDAFEvaluator getEvaluator(ObjectInspector[] parameters) throws UDAFArgumentException {
        // 校验输入参数：只能是一个String类型的字段
        if (parameters.length != 1 || !(parameters[0] instanceof StringObjectInspector)) {
            throw new UDAFArgumentException("UDAF parsed_user_agent only accepts a single String parameter");
        }
        return new ParsedUserAgentEvaluator();
    }

    // 内部类实现具体的聚合逻辑
    public static class ParsedUserAgentEvaluator extends GenericUDAFEvaluator {
        // 定义输入、中间状态、输出的ObjectInspector
        private StringObjectInspector inputUaInspector;
        private StructObjectInspector intermediateInspector;
        private StructObjectInspector finalOutputInspector;

        // 自定义的聚合状态类：用来存三个统计Map
        public static class AggregationState {
            public Map<String, Integer> osMap = new HashMap<>();
            public Map<String, Integer> browserMap = new HashMap<>();
            public Map<String, Integer> deviceMap = new HashMap<>();
        }

        @Override
        public ObjectInspector init(Mode mode, ObjectInspector[] parameters) throws HiveException {
            super.init(mode, parameters);
            switch (mode) {
                case PARTIAL1:
                case PARTIAL2:
                    // 初始化中间状态的ObjectInspector
                    intermediateInspector = ObjectInspectorFactory.getStandardStructObjectInspector(
                        Arrays.asList("os_map", "browser_map", "device_map"),
                        Arrays.asList(
                            ObjectInspectorFactory.getStandardMapObjectInspector(
                                PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                                PrimitiveObjectInspectorFactory.javaIntObjectInspector
                            ),
                            ObjectInspectorFactory.getStandardMapObjectInspector(
                                PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                                PrimitiveObjectInspectorFactory.javaIntObjectInspector
                            ),
                            ObjectInspectorFactory.getStandardMapObjectInspector(
                                PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                                PrimitiveObjectInspectorFactory.javaIntObjectInspector
                            )
                        )
                    );
                    return intermediateInspector;
                case FINAL:
                    // 初始化最终输出的ObjectInspector
                    finalOutputInspector = ObjectInspectorFactory.getStandardStructObjectInspector(
                        Arrays.asList("os_stats", "browser_stats", "device_stats"),
                        Arrays.asList(
                            ObjectInspectorFactory.getStandardMapObjectInspector(
                                PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                                PrimitiveObjectInspectorFactory.javaIntObjectInspector
                            ),
                            ObjectInspectorFactory.getStandardMapObjectInspector(
                                PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                                PrimitiveObjectInspectorFactory.javaIntObjectInspector
                            ),
                            ObjectInspectorFactory.getStandardMapObjectInspector(
                                PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                                PrimitiveObjectInspectorFactory.javaIntObjectInspector
                            )
                        )
                    );
                    return finalOutputInspector;
                case COMPLETE:
                    inputUaInspector = (StringObjectInspector) parameters[0];
                    return finalOutputInspector;
                default:
                    throw new IllegalArgumentException("Unsupported mode: " + mode);
            }
        }

        @Override
        public AggregationState getNewAggregationBuffer() throws HiveException {
            return new AggregationState();
        }

        @Override
        public void reset(AggregationBuffer agg) throws HiveException {
            AggregationState state = (AggregationState) agg;
            state.osMap.clear();
            state.browserMap.clear();
            state.deviceMap.clear();
        }

        @Override
        public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException {
            if (parameters[0] == null) {
                return;
            }
            AggregationState state = (AggregationState) agg;
            String userAgent = inputUaInspector.getPrimitiveJavaObject(parameters[0]);
            
            // 这里调用你已经掌握的User-Agent解析方法，得到os、browser、device
            String os = parseOs(userAgent);
            String browser = parseBrowser(userAgent);
            String device = parseDevice(userAgent);

            // 更新统计Map
            updateMap(state.osMap, os);
            updateMap(state.browserMap, browser);
            updateMap(state.deviceMap, device);
        }

        @Override
        public Object terminatePartial(AggregationBuffer agg) throws HiveException {
            AggregationState state = (AggregationState) agg;
            return new Object[]{state.osMap, state.browserMap, state.deviceMap};
        }

        @Override
        public void merge(AggregationBuffer agg, Object partial) throws HiveException {
            if (partial == null) {
                return;
            }
            AggregationState state = (AggregationState) agg;
            // 从partial中取出各Map
            Map<String, Integer> partialOsMap = (Map<String, Integer>) intermediateInspector.getStructFieldData(partial, intermediateInspector.getStructFieldRef("os_map"));
            Map<String, Integer> partialBrowserMap = (Map<String, Integer>) intermediateInspector.getStructFieldData(partial, intermediateInspector.getStructFieldRef("browser_map"));
            Map<String, Integer> partialDeviceMap = (Map<String, Integer>) intermediateInspector.getStructFieldData(partial, intermediateInspector.getStructFieldRef("device_map"));

            // 合并到全局状态
            mergeMaps(state.osMap, partialOsMap);
            mergeMaps(state.browserMap, partialBrowserMap);
            mergeMaps(state.deviceMap, partialDeviceMap);
        }

        @Override
        public Object terminate(AggregationBuffer agg) throws HiveException {
            AggregationState state = (AggregationState) agg;
            return new Object[]{state.osMap, state.browserMap, state.deviceMap};
        }

        // 工具方法：更新单个Map的计数
        private void updateMap(Map<String, Integer> map, String key) {
            if (key == null || key.isEmpty()) {
                key = "Unknown";
            }
            map.put(key, map.getOrDefault(key, 0) + 1);
        }

        // 工具方法：合并两个统计Map
        private void mergeMaps(Map<String, Integer> target, Map<String, Integer> source) {
            for (Map.Entry<String, Integer> entry : source.entrySet()) {
                target.put(entry.getKey(), target.getOrDefault(entry.getKey(), 0) + entry.getValue());
            }
        }

        // 这里替换成你自己的User-Agent解析实现
        private String parseOs(String userAgent) {
            // 示例逻辑：根据User-Agent字符串提取操作系统
            if (userAgent.contains("Windows")) {
                return "Windows";
            } else if (userAgent.contains("Mac OS")) {
                return "Mac OS";
            } else if (userAgent.contains("Linux")) {
                return "Linux";
            } else {
                return "Unknown";
            }
        }

        // 替换成你的浏览器解析逻辑
        private String parseBrowser(String userAgent) {
            if (userAgent.contains("Chrome")) {
                return "Chrome";
            } else if (userAgent.contains("Firefox")) {
                return "Firefox";
            } else if (userAgent.contains("Googlebot")) {
                return "Googlebot";
            } else {
                return "Unknown";
            }
        }

        // 替换成你的设备解析逻辑
        private String parseDevice(String userAgent) {
            if (userAgent.contains("Mobile")) {
                return "Mobile";
            } else if (userAgent.contains("Tablet")) {
                return "Tablet";
            } else {
                return "Desktop";
            }
        }
    }
}

第二步：编译注册UDAF

把上面的代码编译成JAR包（需要依赖Hive的相关核心包，比如hive-exec、hive-serde等）
在Hive中注册函数：

ADD JAR /path/to/your/udaf.jar;
CREATE FUNCTION parsed_user_agent AS 'com.your.package.ParsedUserAgentUDAF';

第三步：使用UDAF查询

现在就可以执行你想要的分组查询了：

SELECT 
    city,
    parsed_user_agent(User_Agent) AS ua_stats
FROM log_data
GROUP BY city;

如果需要单独提取结构体中的字段，可以用点语法访问：

SELECT 
    city,
    ua_stats.os_stats AS os_distribution,
    ua_stats.browser_stats AS browser_distribution,
    ua_stats.device_stats AS device_distribution
FROM (
    SELECT 
        city,
        parsed_user_agent(User_Agent) AS ua_stats
    FROM log_data
    GROUP BY city
) t;

内容的提问来源于stack exchange，提问作者Pavel Orlov

如何用Java编写继承AbstractGenericUDAFResolver的Hive UDAF函数parsed_user_agent

首先明确输入输出的类型定义 #

第一步：实现继承AbstractGenericUDAFResolver的Resolver类 #

第二步：编译注册UDAF #

第三步：使用UDAF查询 #

首先明确输入输出的类型定义

第一步：实现继承AbstractGenericUDAFResolver的Resolver类

第二步：编译注册UDAF

第三步：使用UDAF查询