1. 项目概述为什么Hive自定义函数是数据工程师的必备技能在数据仓库和离线批处理的世界里Hive SQL是我们最常打交道的语言。但你是否遇到过这样的场景业务方需要一个复杂的字符串解析逻辑或者需要对一列数据进行自定义的聚合统计而内置的concat、sum、avg却怎么也满足不了需求这时候Hive的自定义函数User-Defined Functions, UDFs就成为了破局的关键。它允许我们像使用内置函数一样用Java编写自己的业务逻辑极大地扩展了Hive SQL的表达能力。简单来说Hive UDF就是数据工程师手中的“瑞士军刀”。当标准SQL工具箱里的工具不够用时我们可以自己锻造一把趁手的。根据功能形态的不同这把“军刀”主要分为三类UDF用户自定义函数、UDTF用户自定义表生成函数和UDAF用户自定义聚合函数。理解这三者的区别、适用场景和实现细节是从“会用Hive”到“精通Hive”的重要分水岭。本文将从一个数据开发老兵的实战视角彻底拆解这三类函数不仅告诉你它们是什么更会深入剖析其底层原理、实现步骤并分享那些官方文档里不会写的“踩坑”经验和性能调优技巧。2. Hive自定义函数核心概念与设计思路拆解在动手写代码之前我们必须先厘清核心概念。Hive自定义函数的设计本质上是对MapReduce计算模型中不同阶段计算逻辑的抽象和封装。理解这一点你就能明白为什么会有三种不同的类型以及它们各自应该在什么场景下使用。2.1 三类函数的核心区别与设计哲学UDF (User-Defined Function)这是最基础、最常用的一类。它的设计哲学是“一对一”的映射。你输入一行数据中的一个或多个字段它经过计算后输出一个单一的值。在MapReduce的语境下UDF通常运行在Map阶段或Reduce阶段的单条记录处理环节。例如将一个手机号脱敏138****1234或者将一段JSON字符串解析出某个key对应的value。它的生命周期很短只处理当前这一条记录处理完就释放。UDTF (User-Defined Table-Generating Function)它的设计哲学是“一对多”的爆炸。输入一行数据可以输出零行、一行或多行数据。这是它与UDF最本质的区别。在实现上UDTF通常与LATERAL VIEW语法联用。它的典型场景是“行转列”比如将一行数据中一个包含逗号分隔值的字符串如“苹果,香蕉,橙子”炸开成三行独立的记录。在MapReduce中它也是在Map阶段对单条记录进行操作但输出结果可能改变数据的总行数。UDAF (User-Defined Aggregation Function)这是三类中最复杂、也最体现分布式计算思想的一类。它的设计哲学是“多对一”的聚合。它的输入是多行数据一个分组内的所有数据经过一个复杂的、有状态的计算过程最终输出一个单一的聚合值。sum、count、avg这些内置函数都是UDAF。在MapReduce中UDAF的逻辑贯穿Map端的局部聚合Combiner和Reduce端的全局聚合。因此实现一个UDAF你需要清晰地定义如何初始化一个聚合缓冲区、如何迭代更新这个缓冲区、以及如何合并来自不同Map任务的局部聚合结果。注意很多初学者容易混淆UDAF和“在UDF里做循环聚合”。请牢记UDAF是Hive框架在分布式环境下帮你管理聚合状态和过程的而用UDF硬写聚合逻辑不仅代码复杂而且无法利用Hive的优化器性能会非常差。2.2 技术选型背后的考量何时该用哪一种选择哪种函数取决于你的输入和输出形态。当你需要对单行数据进行转换或计算且输出是单个值时用UDF。这是最直观的选择。当你需要将单行数据拆成多行或者生成一个虚拟表与原表进行连接时用UDTF。典型场景是解析数组、Map类型的字段或者做数据探查例如生成一个序列。当你需要对一组行一个窗口或一个分组进行统计计算时必须用UDAF。任何涉及GROUP BY的复杂聚合逻辑都是UDAF的用武之地。从开发复杂度上看UDF最简单UDTF次之UDAF最复杂。但复杂度也带来了更强的能力。理解这个选型逻辑能让你在项目初期就做出正确的技术决策避免后期重构。3. 核心细节解析与实操要点了解了宏观概念我们深入到每一类函数的实现细节。这里会包含大量的代码示例和关键注解这些注解正是从无数个线上任务调试中总结出的“血泪经验”。3.1 UDF实现详解从Hello World到复杂逻辑一个最简单的UDF就是继承Hive提供的org.apache.hadoop.hive.ql.exec.UDF类并重写evaluate方法。这个方法支持重载你可以定义多个不同参数类型的evaluate方法Hive会根据调用时的参数类型自动匹配。import org.apache.hadoop.hive.ql.exec.UDF; import org.apache.hadoop.io.Text; public class SimpleUDFExample extends UDF { // 方法名必须是 evaluate public Text evaluate(Text input) { if (input null) { return null; // 处理空值至关重要 } String str input.toString(); // 示例将字符串转换为大写 return new Text(str.toUpperCase()); } // 支持重载处理整数输入 public Text evaluate(Text input, IntWritable times) { if (input null || times null) { return null; } StringBuilder sb new StringBuilder(); for (int i 0; i times.get(); i) { sb.append(input.toString()); } return new Text(sb.toString()); } }实操要点与避坑指南空值处理是第一要务生产环境的数据永远是不干净的。你的evaluate方法必须能够优雅地处理null输入并返回null或其他默认值。否则一个NullPointerException会导致整个Map或Reduce任务失败。使用Hadoop Writable类型注意方法的参数和返回值类型推荐使用Text、IntWritable、DoubleWritable等Hadoop的Writable类型而不是Java原生的String、int、double。这是因为Hive在序列化和反序列化数据时默认使用这些类型以获得更好的性能。虽然Hive有类型转换机制但直接使用Writable类型更安全、更高效。避免在UDF中创建大量临时对象evaluate方法会被海量数据调用无数次。如果在方法内部频繁创建new Text()或new StringBuilder()会引发大量的垃圾回收GC严重拖慢任务速度。一个常见的优化是对于简单的字符串操作可以考虑复用对象但要注意线程安全通常UDF实例是线程安全的但具体看Hive版本和配置。3.2 UDTF实现详解掌握“一拆多”的艺术UDTF需要继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF类。你需要实现三个关键方法initialize: 初始化定义输出数据的列名和类型。process: 核心处理逻辑输入一行数据通过forward方法输出零行或多行结果。close: 清理资源。import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF; import org.apache.hadoop.hive.ql.exec.UDFArgumentException; import org.apache.hadoop.hive.ql.metadata.HiveException; import org.apache.hadoop.hive.serde2.objectinspector.*; import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory; import java.util.ArrayList; public class SplitStringUDTF extends GenericUDTF { // 声明输出的列名和类型检查器 private PrimitiveObjectInspector stringOI null; Override public StructObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException { // 1. 检查参数个数和类型 if (args.length ! 1) { throw new UDFArgumentException(SplitStringUDTF() takes exactly one argument); } if (args[0].getCategory() ! ObjectInspector.Category.PRIMITIVE) { throw new UDFArgumentException(SplitStringUDTF() requires a primitive argument); } stringOI (PrimitiveObjectInspector) args[0]; if (stringOI.getPrimitiveCategory() ! PrimitiveObjectInspector.PrimitiveCategory.STRING) { throw new UDFArgumentException(SplitStringUDTF() requires a string argument); } // 2. 定义输出列名和类型 ArrayListString fieldNames new ArrayListString(); ArrayListObjectInspector fieldOIs new ArrayListObjectInspector(); fieldNames.add(split_item); fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector); // 输出类型为String return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs); } Override public void process(Object[] record) throws HiveException { // 获取输入字符串 String input stringOI.getPrimitiveJavaObject(record[0]).toString(); if (input null || input.isEmpty()) { return; // 输入为空不输出任何行 } // 按逗号分割 String[] items input.split(,); for (String item : items) { // 3. 关键通过forward逐行输出 forward(new Object[]{item.trim()}); // 注意去空格 } } Override public void close() throws HiveException { // 这里可以释放资源如关闭文件流、数据库连接等。 // 本例无资源需要释放。 } }UDTF的核心难点与技巧initialize方法中的ObjectInspector这是Hive用来解构和访问复杂数据对象的“镜子”系统。对于初学者这块最让人头疼。简单理解它告诉Hive你的函数输入参数是什么类型以及你打算输出什么结构的数据。上面的例子中我们检查输入是一个基本类型Primitive的字符串并声明输出是一个名为split_item的字符串列。forward方法的调用process方法里每调用一次forward就产生一行输出数据。参数是一个Object[]数组其长度和类型必须与initialize中声明的输出结构完全一致。与LATERAL VIEW的配合UDTF很少单独使用必须结合LATERAL VIEW语法。例如SELECT pageid, adid FROM page_ads LATERAL VIEW explode(adid_list) adTable AS adid;其中explode就是一个内置的UDTF。我们自定义的UDTF用法类似。3.3 UDAF实现详解理解分布式聚合的生命周期UDAF的实现最为复杂通常继承org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator的内部类AbstractAggregationBuffer来管理聚合状态并实现一系列生命周期方法。更现代、更推荐的方式是使用Hive 2.3.0之后引入的**GenericUDAFResolver2接口和注解方式**这大大简化了实现。这里我们以计算一组数据平均值的UDAF为例展示推荐的新方式。首先定义一个存储中间状态的Buffer类import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator; public class AvgBuffer extends GenericUDAFEvaluator.AbstractAggregationBuffer { private long count; // 记录数量 private double sum; // 记录总和 // ... 相应的getter和setter方法 }然后实现核心的Evaluator类。一个完整的UDAF需要处理聚合的多个模式ModePARTIAL1 (Map阶段): 从原始数据到局部聚合。对应iterate和terminatePartial。PARTIAL2 (Combine阶段): 合并局部聚合结果。对应merge和terminatePartial。FINAL (Reduce阶段): 生成最终结果。对应merge和terminate。COMPLETE (如果只有Map阶段): 从原始数据直接到最终结果。Description(name my_avg, value _FUNC_(x) - Returns the average of a set of numbers) public class GenericUDAFMyAvg extends AbstractGenericUDAFResolver { Override public GenericUDAFEvaluator getEvaluator(GenericUDAFParameterInfo info) throws SemanticException { // 检查参数类型等 return new GenericUDAFMyAvgEvaluator(); } public static class GenericUDAFMyAvgEvaluator extends GenericUDAFEvaluator { // 声明输入、中间结果、最终结果的类型检查器 private PrimitiveObjectInspector inputOI; private StandardListObjectInspector listOI; // 中间结果用List存储[sum, count] private DoubleObjectInspector outputOI; // 定义聚合缓冲区 static class AvgAggBuffer implements AggregationBuffer { long count; double sum; } Override public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException { super.init(m, parameters); // 根据不同的Mode初始化不同的ObjectInspector if (m Mode.PARTIAL1 || m Mode.COMPLETE) { // 原始输入是double inputOI (PrimitiveObjectInspector) parameters[0]; return ObjectInspectorFactory.getStandardListObjectInspector( PrimitiveObjectInspectorFactory.writableDoubleObjectInspector); } else if (m Mode.PARTIAL2 || m Mode.FINAL) { // 中间输入是ListDoubleWritable listOI (StandardListObjectInspector) parameters[0]; return ObjectInspectorFactory.getStandardListObjectInspector( PrimitiveObjectInspectorFactory.writableDoubleObjectInspector); } else { // Mode.FINAL // 最终输出是double return PrimitiveObjectInspectorFactory.writableDoubleObjectInspector; } } Override public AggregationBuffer getNewAggregationBuffer() throws HiveException { AvgAggBuffer buffer new AvgAggBuffer(); reset(buffer); return buffer; } Override public void reset(AggregationBuffer agg) throws HiveException { ((AvgAggBuffer) agg).count 0; ((AvgAggBuffer) agg).sum 0; } // 迭代处理一条新数据 Override public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException { if (parameters[0] null) return; // 忽略空值 double value PrimitiveObjectInspectorUtils.getDouble(parameters[0], inputOI); ((AvgAggBuffer) agg).sum value; ((AvgAggBuffer) agg).count; } // 终止局部聚合返回中间结果 Override public Object terminatePartial(AggregationBuffer agg) throws HiveException { AvgAggBuffer buffer (AvgAggBuffer) agg; ListDoubleWritable result new ArrayList(2); result.add(new DoubleWritable(buffer.sum)); result.add(new DoubleWritable(buffer.count)); return result; } // 合并合并两个局部聚合结果 Override public void merge(AggregationBuffer agg, Object partial) throws HiveException { if (partial null) return; ListDoubleWritable list (ListDoubleWritable) listOI.getList(partial); double otherSum list.get(0).get(); long otherCount (long) list.get(1).get(); AvgAggBuffer buffer (AvgAggBuffer) agg; buffer.sum otherSum; buffer.count otherCount; } // 终止返回最终聚合结果 Override public Object terminate(AggregationBuffer agg) throws HiveException { AvgAggBuffer buffer (AvgAggBuffer) agg; if (buffer.count 0) { return null; // 没有数据返回null } return new DoubleWritable(buffer.sum / buffer.count); } } }UDAF实现的心得体会深刻理解Mode模式这是实现UDAF最难也是最重要的部分。你必须清晰地知道你的代码在聚合的哪个阶段Map、Combine、Reduce被调用以及当前输入和输出的数据类型是什么。上面的init方法根据不同的Mode返回不同的ObjectInspector就是为此服务。中间结果的设计terminatePartial返回的中间结果必须能被merge方法正确解析。通常使用数组List或结构体来同时传递多个聚合状态如总和与计数。设计良好的中间结果格式是保证分布式聚合正确性的关键。性能考虑iterate和merge方法会被调用极其频繁。里面的逻辑要尽可能高效避免复杂的对象创建和拆箱装箱操作。对于数值类型直接使用基本类型运算。4. 完整实操流程从开发到上线理论说得再多不如亲手跑一遍。下面我们以一个完整的UDF开发部署流程为例串联起所有环节。4.1 环境准备与项目搭建假设我们使用Maven管理项目。在你的pom.xml中需要引入Hive的执行引擎依赖。注意依赖的Hive版本必须与线上集群的版本一致这是避免出现ClassNotFoundException或方法签名不匹配问题的首要原则。dependency groupIdorg.apache.hive/groupId artifactIdhive-exec/artifactId version2.3.9/version !-- 请替换为你的集群版本 -- scopeprovided/scope !-- 因为集群上已有打包时不需要包含 -- /dependency dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-common/artifactId version2.7.7/version !-- 匹配Hadoop版本 -- scopeprovided/scope /dependency使用scope为provided是因为这些Jar包在Hive服务器上已经存在我们只需要在编译时使用它们最终打出的UDF Jar包不应该包含它们否则可能引起版本冲突。4.2 编写、打包与上传编写Java类如上节示例在src/main/java下创建你的UDF类。打包在项目根目录下执行mvn clean package。这会在target目录下生成一个类似my-hive-udfs-1.0-SNAPSHOT.jar的文件。上传至HDFS这是关键一步。为了让所有HiveServer节点都能访问到你的UDF Jar包必须将其上传到分布式文件系统如HDFS。hadoop fs -mkdir -p /lib/hive/udfs/ # 创建目录 hadoop fs -put target/my-hive-udfs-1.0-SNAPSHOT.jar /lib/hive/udfs/4.3 Hive会话中注册与使用连接到Hive通过Beeline或Hive CLI执行以下命令-- 1. 将Jar包添加到本次会话的类路径中 ADD JAR hdfs:///lib/hive/udfs/my-hive-udfs-1.0-SNAPSHOT.jar; -- 2. 创建临时函数仅本次会话有效 CREATE TEMPORARY FUNCTION my_upper AS com.yourcompany.hive.udf.SimpleUDFExample; CREATE TEMPORARY FUNCTION my_split AS com.yourcompany.hive.udtf.SplitStringUDTF; CREATE TEMPORARY FUNCTION my_avg AS com.yourcompany.hive.udaf.GenericUDAFMyAvg; -- 3. 使用函数 SELECT my_upper(username) FROM user_table; SELECT pageid, item FROM page_table LATERAL VIEW my_split(tags) t AS item; SELECT category, my_avg(price) FROM sales_table GROUP BY category;临时函数与永久函数临时函数使用CREATE TEMPORARY FUNCTION。它的生命周期仅限于当前Hive会话。断开重连后函数就消失了。适合临时测试和探索。永久函数使用CREATE FUNCTION。函数元数据会存储在Hive Metastore中永久有效。创建永久函数时Jar包位置必须使用HDFS路径。CREATE FUNCTION my_permanent_avg AS com.yourcompany.hive.udaf.GenericUDAFMyAvg USING JAR hdfs:///lib/hive/udfs/my-hive-udfs-1.0-SNAPSHOT.jar;永久函数对所有用户和会话都可用是生产环境的标准做法。4.4 实操现场记录一个复杂的JSON解析UDF让我们看一个更贴近生产的例子解析用户行为日志中的JSON字段。日志中有一个extra_info字段是JSON字符串我们需要从中提取device_model和app_version。import org.apache.hadoop.hive.ql.exec.UDF; import org.apache.hadoop.io.Text; import org.json.JSONObject; // 可以使用org.json库 import org.json.JSONException; public class ParseJsonUDF extends UDF { private Text result new Text(); // 复用对象减少GC public Text evaluate(Text jsonStr, Text key) { if (jsonStr null || key null) { return null; } try { JSONObject json new JSONObject(jsonStr.toString()); if (json.has(key.toString())) { result.set(json.getString(key.toString())); return result; } else { return null; // key不存在 } } catch (JSONException e) { // 记录解析错误但不要抛出异常导致任务失败返回null // 在实际生产中这里可以增加日志输出便于排查脏数据 return null; } } }使用方式ADD JAR /path/to/json-lib.jar; -- 别忘了添加org.json库的Jar包 ADD JAR /path/to/your-udf.jar; CREATE TEMPORARY FUNCTION json_get AS com.xxx.ParseJsonUDF; SELECT user_id, json_get(extra_info, device_model) as device, json_get(extra_info, app_version) as version FROM user_log_table;这个例子展示了生产级UDF的几个要点健壮的空值和异常处理、第三方库的依赖管理、以及通过复用对象来优化性能。5. 常见问题、排查技巧与性能优化实录即使代码写对了在部署和使用过程中你依然会遇到各种各样的问题。下面是我在多年运维中积累的一些典型问题及其解决方案。5.1 常见问题速查表问题现象可能原因排查步骤与解决方案ClassNotFoundException或NoClassDefFoundError1. Jar包未正确添加到会话。2. Jar包中缺少依赖。3. Hive Server的classpath配置问题。1. 确认ADD JAR命令执行成功且路径正确HDFS路径需有权限。2. 使用mvn dependency:tree检查并打包所有非provided依赖到UDF Jar生成fat jar或使用ADD JAR依次添加所有依赖Jar。3. 联系集群管理员确认Hive Server的hive.aux.jars.path配置是否包含常用UDF路径。FAILED: SemanticException [Error 10011]: Invalid function1. 函数名重复或冲突。2. 创建函数时指定的类名错误。1. 使用SHOW FUNCTIONS LIKE *your_func*;查看是否已存在同名函数。临时函数和永久函数是分开的命名空间。2. 仔细检查CREATE FUNCTION语句中的全限定类名确保与Jar包中的类路径完全一致。UDF返回结果全是NULL1. UDF代码逻辑中未处理输入为null的情况直接返回null。2. 数据类型不匹配Hive进行了隐式转换失败。3. 业务逻辑本身导致无输出。1. 在UDF的evaluate方法开始处增加日志打印输入参数确认数据是否正常传入。2. 检查Hive表中字段类型与UDF方法声明的参数类型Text,IntWritable等是否兼容。3. 简化UDF逻辑先写一个返回固定值的版本进行测试排除业务代码问题。UDTF与LATERAL VIEW联用时报错或结果不对1. UDTF输出的列数与LATERAL VIEW ... AS后面指定的别名数量不匹配。2. UDTF的forward方法输出的对象类型与initialize声明的类型不一致。1. 确认initialize方法中定义的输出列数量fieldNames的size与AS后的别名数量一致。2. 在forward方法中打断点或打印日志确认每次输出的Object[]数组长度和内容是否符合预期。UDAF在分布式运行时结果错误1.merge方法逻辑错误合并状态时出错。2. 中间结果terminatePartial返回值序列化/反序列化有问题。3. 聚合缓冲区AggregationBuffer的reset方法未正确初始化。1.这是最棘手的。首先在本地模式下set hive.exec.mode.local.autotrue;测试小数据集结果正确后再测分布式。2. 确保terminatePartial返回的对象能被对应的ObjectInspector正确解析。对于复杂对象考虑使用Hive可序列化的标准类型如ArrayListDoubleWritable。3. 在getNewAggregationBuffer和reset方法中确保所有状态变量都被初始化。性能极差任务运行缓慢1. UDF/UDTF/UDAF内部有耗资源操作如频繁创建大对象、正则表达式编译、网络IO。2. 数据倾斜某些键Key对应的数据量巨大。1.Profile你的代码。避免在evaluate、process、iterate等方法内做重复初始化如Pattern.compile应放在类初始化阶段。强烈复用对象。2. 对于UDAF检查是否因某个分组数据量过大导致单个Reducer卡住。尝试通过set hive.groupby.skewindatatrue;开启倾斜优化或对数据先进行预处理。5.2 性能优化独家心得对象复用是黄金法则在UDF的evaluate方法中声明一个成员变量private Text result new Text();然后在方法内result.set(...); return result;。这能减少海量调用中产生的垃圾对象对性能提升立竿见影。对于UDTF和UDAF也要注意在forward或返回结果时尽量复用对象数组。谨慎使用复杂第三方库像org.json这样的库虽然方便但可能比较重。如果只是解析简单的JSON路径可以考虑使用更轻量级的库如Jackson或Gson甚至自己写简单的字符串解析。务必在打包时处理好依赖。利用Hive参数进行调试set hive.udtf.auto.progressfalse;可以关闭UDTF的进度报告有时能解决一些进度卡住的问题。set hive.exec.paralleltrue;开启任务并行对于多个UDF/UDTF阶段的任务有加速效果。对于UDAF可以通过set hive.map.aggrtrue;默认开启在Map端进行聚合减少Shuffle数据量。永久函数的管理生产环境建议建立规范的UDF管理流程。例如将所有的UDF Jar包统一上传到HDFS的特定目录如/data/udf_libs/并使用统一的命名规范。创建函数的SQL脚本纳入版本管理如Git。当UDF更新时需要先DROP FUNCTION再ADD JAR新版本最后CREATE FUNCTION。注意这可能会影响正在运行或依赖该函数的作业最好在业务低峰期操作。5.3 调试技巧如何看到UDF内部的日志这是新手最常问的问题。UDF运行在分布式集群的YARN容器里如何打印和查看日志使用System.err.println这是最直接的方法。在UDF代码中打印的信息会输出到该任务容器的标准错误stderr日志中。查看YARN日志首先在Hive CLI或Beeline中找到你的应用IDapplication_xxx_xxxx。通过YARN ResourceManager的Web UI通常8088端口找到该应用。点击应用进入“ApplicationMaster”的日志或者直接查看各个Map/Reduce Task的“Container Logs”。在Container日志里找到stderr文件你就能看到System.err.println输出的内容了。集成SLF4J日志框架对于更复杂的日志管理可以在UDF项目中引入slf4j-api和log4j等依赖并配置日志文件。但要注意日志文件会写在容器本地任务结束后会被清理需要配置日志聚合到HDFS才能长期查看。最后我个人最深刻的体会是自定义函数是Hive能力的延伸但它也是一把双刃剑。滥用UDF特别是低效的UDF会严重拖慢整个集群的任务速度。在决定自己写UDF之前务必先查一查Hive的内置函数是否已经能满足需求。如果非要写一定要把性能、健壮性空值、异常处理和可维护性清晰的命名和注释放在首位。将通用的、稳定的UDF固化下来形成团队的函数库能极大提升数据开发的效率和质量。 Hive UDF/UDTF/UDAF：从核心原理到生产级实现与调优
1. 项目概述为什么Hive自定义函数是数据工程师的必备技能在数据仓库和离线批处理的世界里Hive SQL是我们最常打交道的语言。但你是否遇到过这样的场景业务方需要一个复杂的字符串解析逻辑或者需要对一列数据进行自定义的聚合统计而内置的concat、sum、avg却怎么也满足不了需求这时候Hive的自定义函数User-Defined Functions, UDFs就成为了破局的关键。它允许我们像使用内置函数一样用Java编写自己的业务逻辑极大地扩展了Hive SQL的表达能力。简单来说Hive UDF就是数据工程师手中的“瑞士军刀”。当标准SQL工具箱里的工具不够用时我们可以自己锻造一把趁手的。根据功能形态的不同这把“军刀”主要分为三类UDF用户自定义函数、UDTF用户自定义表生成函数和UDAF用户自定义聚合函数。理解这三者的区别、适用场景和实现细节是从“会用Hive”到“精通Hive”的重要分水岭。本文将从一个数据开发老兵的实战视角彻底拆解这三类函数不仅告诉你它们是什么更会深入剖析其底层原理、实现步骤并分享那些官方文档里不会写的“踩坑”经验和性能调优技巧。2. Hive自定义函数核心概念与设计思路拆解在动手写代码之前我们必须先厘清核心概念。Hive自定义函数的设计本质上是对MapReduce计算模型中不同阶段计算逻辑的抽象和封装。理解这一点你就能明白为什么会有三种不同的类型以及它们各自应该在什么场景下使用。2.1 三类函数的核心区别与设计哲学UDF (User-Defined Function)这是最基础、最常用的一类。它的设计哲学是“一对一”的映射。你输入一行数据中的一个或多个字段它经过计算后输出一个单一的值。在MapReduce的语境下UDF通常运行在Map阶段或Reduce阶段的单条记录处理环节。例如将一个手机号脱敏138****1234或者将一段JSON字符串解析出某个key对应的value。它的生命周期很短只处理当前这一条记录处理完就释放。UDTF (User-Defined Table-Generating Function)它的设计哲学是“一对多”的爆炸。输入一行数据可以输出零行、一行或多行数据。这是它与UDF最本质的区别。在实现上UDTF通常与LATERAL VIEW语法联用。它的典型场景是“行转列”比如将一行数据中一个包含逗号分隔值的字符串如“苹果,香蕉,橙子”炸开成三行独立的记录。在MapReduce中它也是在Map阶段对单条记录进行操作但输出结果可能改变数据的总行数。UDAF (User-Defined Aggregation Function)这是三类中最复杂、也最体现分布式计算思想的一类。它的设计哲学是“多对一”的聚合。它的输入是多行数据一个分组内的所有数据经过一个复杂的、有状态的计算过程最终输出一个单一的聚合值。sum、count、avg这些内置函数都是UDAF。在MapReduce中UDAF的逻辑贯穿Map端的局部聚合Combiner和Reduce端的全局聚合。因此实现一个UDAF你需要清晰地定义如何初始化一个聚合缓冲区、如何迭代更新这个缓冲区、以及如何合并来自不同Map任务的局部聚合结果。注意很多初学者容易混淆UDAF和“在UDF里做循环聚合”。请牢记UDAF是Hive框架在分布式环境下帮你管理聚合状态和过程的而用UDF硬写聚合逻辑不仅代码复杂而且无法利用Hive的优化器性能会非常差。2.2 技术选型背后的考量何时该用哪一种选择哪种函数取决于你的输入和输出形态。当你需要对单行数据进行转换或计算且输出是单个值时用UDF。这是最直观的选择。当你需要将单行数据拆成多行或者生成一个虚拟表与原表进行连接时用UDTF。典型场景是解析数组、Map类型的字段或者做数据探查例如生成一个序列。当你需要对一组行一个窗口或一个分组进行统计计算时必须用UDAF。任何涉及GROUP BY的复杂聚合逻辑都是UDAF的用武之地。从开发复杂度上看UDF最简单UDTF次之UDAF最复杂。但复杂度也带来了更强的能力。理解这个选型逻辑能让你在项目初期就做出正确的技术决策避免后期重构。3. 核心细节解析与实操要点了解了宏观概念我们深入到每一类函数的实现细节。这里会包含大量的代码示例和关键注解这些注解正是从无数个线上任务调试中总结出的“血泪经验”。3.1 UDF实现详解从Hello World到复杂逻辑一个最简单的UDF就是继承Hive提供的org.apache.hadoop.hive.ql.exec.UDF类并重写evaluate方法。这个方法支持重载你可以定义多个不同参数类型的evaluate方法Hive会根据调用时的参数类型自动匹配。import org.apache.hadoop.hive.ql.exec.UDF; import org.apache.hadoop.io.Text; public class SimpleUDFExample extends UDF { // 方法名必须是 evaluate public Text evaluate(Text input) { if (input null) { return null; // 处理空值至关重要 } String str input.toString(); // 示例将字符串转换为大写 return new Text(str.toUpperCase()); } // 支持重载处理整数输入 public Text evaluate(Text input, IntWritable times) { if (input null || times null) { return null; } StringBuilder sb new StringBuilder(); for (int i 0; i times.get(); i) { sb.append(input.toString()); } return new Text(sb.toString()); } }实操要点与避坑指南空值处理是第一要务生产环境的数据永远是不干净的。你的evaluate方法必须能够优雅地处理null输入并返回null或其他默认值。否则一个NullPointerException会导致整个Map或Reduce任务失败。使用Hadoop Writable类型注意方法的参数和返回值类型推荐使用Text、IntWritable、DoubleWritable等Hadoop的Writable类型而不是Java原生的String、int、double。这是因为Hive在序列化和反序列化数据时默认使用这些类型以获得更好的性能。虽然Hive有类型转换机制但直接使用Writable类型更安全、更高效。避免在UDF中创建大量临时对象evaluate方法会被海量数据调用无数次。如果在方法内部频繁创建new Text()或new StringBuilder()会引发大量的垃圾回收GC严重拖慢任务速度。一个常见的优化是对于简单的字符串操作可以考虑复用对象但要注意线程安全通常UDF实例是线程安全的但具体看Hive版本和配置。3.2 UDTF实现详解掌握“一拆多”的艺术UDTF需要继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF类。你需要实现三个关键方法initialize: 初始化定义输出数据的列名和类型。process: 核心处理逻辑输入一行数据通过forward方法输出零行或多行结果。close: 清理资源。import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF; import org.apache.hadoop.hive.ql.exec.UDFArgumentException; import org.apache.hadoop.hive.ql.metadata.HiveException; import org.apache.hadoop.hive.serde2.objectinspector.*; import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory; import java.util.ArrayList; public class SplitStringUDTF extends GenericUDTF { // 声明输出的列名和类型检查器 private PrimitiveObjectInspector stringOI null; Override public StructObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException { // 1. 检查参数个数和类型 if (args.length ! 1) { throw new UDFArgumentException(SplitStringUDTF() takes exactly one argument); } if (args[0].getCategory() ! ObjectInspector.Category.PRIMITIVE) { throw new UDFArgumentException(SplitStringUDTF() requires a primitive argument); } stringOI (PrimitiveObjectInspector) args[0]; if (stringOI.getPrimitiveCategory() ! PrimitiveObjectInspector.PrimitiveCategory.STRING) { throw new UDFArgumentException(SplitStringUDTF() requires a string argument); } // 2. 定义输出列名和类型 ArrayListString fieldNames new ArrayListString(); ArrayListObjectInspector fieldOIs new ArrayListObjectInspector(); fieldNames.add(split_item); fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector); // 输出类型为String return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs); } Override public void process(Object[] record) throws HiveException { // 获取输入字符串 String input stringOI.getPrimitiveJavaObject(record[0]).toString(); if (input null || input.isEmpty()) { return; // 输入为空不输出任何行 } // 按逗号分割 String[] items input.split(,); for (String item : items) { // 3. 关键通过forward逐行输出 forward(new Object[]{item.trim()}); // 注意去空格 } } Override public void close() throws HiveException { // 这里可以释放资源如关闭文件流、数据库连接等。 // 本例无资源需要释放。 } }UDTF的核心难点与技巧initialize方法中的ObjectInspector这是Hive用来解构和访问复杂数据对象的“镜子”系统。对于初学者这块最让人头疼。简单理解它告诉Hive你的函数输入参数是什么类型以及你打算输出什么结构的数据。上面的例子中我们检查输入是一个基本类型Primitive的字符串并声明输出是一个名为split_item的字符串列。forward方法的调用process方法里每调用一次forward就产生一行输出数据。参数是一个Object[]数组其长度和类型必须与initialize中声明的输出结构完全一致。与LATERAL VIEW的配合UDTF很少单独使用必须结合LATERAL VIEW语法。例如SELECT pageid, adid FROM page_ads LATERAL VIEW explode(adid_list) adTable AS adid;其中explode就是一个内置的UDTF。我们自定义的UDTF用法类似。3.3 UDAF实现详解理解分布式聚合的生命周期UDAF的实现最为复杂通常继承org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator的内部类AbstractAggregationBuffer来管理聚合状态并实现一系列生命周期方法。更现代、更推荐的方式是使用Hive 2.3.0之后引入的**GenericUDAFResolver2接口和注解方式**这大大简化了实现。这里我们以计算一组数据平均值的UDAF为例展示推荐的新方式。首先定义一个存储中间状态的Buffer类import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator; public class AvgBuffer extends GenericUDAFEvaluator.AbstractAggregationBuffer { private long count; // 记录数量 private double sum; // 记录总和 // ... 相应的getter和setter方法 }然后实现核心的Evaluator类。一个完整的UDAF需要处理聚合的多个模式ModePARTIAL1 (Map阶段): 从原始数据到局部聚合。对应iterate和terminatePartial。PARTIAL2 (Combine阶段): 合并局部聚合结果。对应merge和terminatePartial。FINAL (Reduce阶段): 生成最终结果。对应merge和terminate。COMPLETE (如果只有Map阶段): 从原始数据直接到最终结果。Description(name my_avg, value _FUNC_(x) - Returns the average of a set of numbers) public class GenericUDAFMyAvg extends AbstractGenericUDAFResolver { Override public GenericUDAFEvaluator getEvaluator(GenericUDAFParameterInfo info) throws SemanticException { // 检查参数类型等 return new GenericUDAFMyAvgEvaluator(); } public static class GenericUDAFMyAvgEvaluator extends GenericUDAFEvaluator { // 声明输入、中间结果、最终结果的类型检查器 private PrimitiveObjectInspector inputOI; private StandardListObjectInspector listOI; // 中间结果用List存储[sum, count] private DoubleObjectInspector outputOI; // 定义聚合缓冲区 static class AvgAggBuffer implements AggregationBuffer { long count; double sum; } Override public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException { super.init(m, parameters); // 根据不同的Mode初始化不同的ObjectInspector if (m Mode.PARTIAL1 || m Mode.COMPLETE) { // 原始输入是double inputOI (PrimitiveObjectInspector) parameters[0]; return ObjectInspectorFactory.getStandardListObjectInspector( PrimitiveObjectInspectorFactory.writableDoubleObjectInspector); } else if (m Mode.PARTIAL2 || m Mode.FINAL) { // 中间输入是ListDoubleWritable listOI (StandardListObjectInspector) parameters[0]; return ObjectInspectorFactory.getStandardListObjectInspector( PrimitiveObjectInspectorFactory.writableDoubleObjectInspector); } else { // Mode.FINAL // 最终输出是double return PrimitiveObjectInspectorFactory.writableDoubleObjectInspector; } } Override public AggregationBuffer getNewAggregationBuffer() throws HiveException { AvgAggBuffer buffer new AvgAggBuffer(); reset(buffer); return buffer; } Override public void reset(AggregationBuffer agg) throws HiveException { ((AvgAggBuffer) agg).count 0; ((AvgAggBuffer) agg).sum 0; } // 迭代处理一条新数据 Override public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException { if (parameters[0] null) return; // 忽略空值 double value PrimitiveObjectInspectorUtils.getDouble(parameters[0], inputOI); ((AvgAggBuffer) agg).sum value; ((AvgAggBuffer) agg).count; } // 终止局部聚合返回中间结果 Override public Object terminatePartial(AggregationBuffer agg) throws HiveException { AvgAggBuffer buffer (AvgAggBuffer) agg; ListDoubleWritable result new ArrayList(2); result.add(new DoubleWritable(buffer.sum)); result.add(new DoubleWritable(buffer.count)); return result; } // 合并合并两个局部聚合结果 Override public void merge(AggregationBuffer agg, Object partial) throws HiveException { if (partial null) return; ListDoubleWritable list (ListDoubleWritable) listOI.getList(partial); double otherSum list.get(0).get(); long otherCount (long) list.get(1).get(); AvgAggBuffer buffer (AvgAggBuffer) agg; buffer.sum otherSum; buffer.count otherCount; } // 终止返回最终聚合结果 Override public Object terminate(AggregationBuffer agg) throws HiveException { AvgAggBuffer buffer (AvgAggBuffer) agg; if (buffer.count 0) { return null; // 没有数据返回null } return new DoubleWritable(buffer.sum / buffer.count); } } }UDAF实现的心得体会深刻理解Mode模式这是实现UDAF最难也是最重要的部分。你必须清晰地知道你的代码在聚合的哪个阶段Map、Combine、Reduce被调用以及当前输入和输出的数据类型是什么。上面的init方法根据不同的Mode返回不同的ObjectInspector就是为此服务。中间结果的设计terminatePartial返回的中间结果必须能被merge方法正确解析。通常使用数组List或结构体来同时传递多个聚合状态如总和与计数。设计良好的中间结果格式是保证分布式聚合正确性的关键。性能考虑iterate和merge方法会被调用极其频繁。里面的逻辑要尽可能高效避免复杂的对象创建和拆箱装箱操作。对于数值类型直接使用基本类型运算。4. 完整实操流程从开发到上线理论说得再多不如亲手跑一遍。下面我们以一个完整的UDF开发部署流程为例串联起所有环节。4.1 环境准备与项目搭建假设我们使用Maven管理项目。在你的pom.xml中需要引入Hive的执行引擎依赖。注意依赖的Hive版本必须与线上集群的版本一致这是避免出现ClassNotFoundException或方法签名不匹配问题的首要原则。dependency groupIdorg.apache.hive/groupId artifactIdhive-exec/artifactId version2.3.9/version !-- 请替换为你的集群版本 -- scopeprovided/scope !-- 因为集群上已有打包时不需要包含 -- /dependency dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-common/artifactId version2.7.7/version !-- 匹配Hadoop版本 -- scopeprovided/scope /dependency使用scope为provided是因为这些Jar包在Hive服务器上已经存在我们只需要在编译时使用它们最终打出的UDF Jar包不应该包含它们否则可能引起版本冲突。4.2 编写、打包与上传编写Java类如上节示例在src/main/java下创建你的UDF类。打包在项目根目录下执行mvn clean package。这会在target目录下生成一个类似my-hive-udfs-1.0-SNAPSHOT.jar的文件。上传至HDFS这是关键一步。为了让所有HiveServer节点都能访问到你的UDF Jar包必须将其上传到分布式文件系统如HDFS。hadoop fs -mkdir -p /lib/hive/udfs/ # 创建目录 hadoop fs -put target/my-hive-udfs-1.0-SNAPSHOT.jar /lib/hive/udfs/4.3 Hive会话中注册与使用连接到Hive通过Beeline或Hive CLI执行以下命令-- 1. 将Jar包添加到本次会话的类路径中 ADD JAR hdfs:///lib/hive/udfs/my-hive-udfs-1.0-SNAPSHOT.jar; -- 2. 创建临时函数仅本次会话有效 CREATE TEMPORARY FUNCTION my_upper AS com.yourcompany.hive.udf.SimpleUDFExample; CREATE TEMPORARY FUNCTION my_split AS com.yourcompany.hive.udtf.SplitStringUDTF; CREATE TEMPORARY FUNCTION my_avg AS com.yourcompany.hive.udaf.GenericUDAFMyAvg; -- 3. 使用函数 SELECT my_upper(username) FROM user_table; SELECT pageid, item FROM page_table LATERAL VIEW my_split(tags) t AS item; SELECT category, my_avg(price) FROM sales_table GROUP BY category;临时函数与永久函数临时函数使用CREATE TEMPORARY FUNCTION。它的生命周期仅限于当前Hive会话。断开重连后函数就消失了。适合临时测试和探索。永久函数使用CREATE FUNCTION。函数元数据会存储在Hive Metastore中永久有效。创建永久函数时Jar包位置必须使用HDFS路径。CREATE FUNCTION my_permanent_avg AS com.yourcompany.hive.udaf.GenericUDAFMyAvg USING JAR hdfs:///lib/hive/udfs/my-hive-udfs-1.0-SNAPSHOT.jar;永久函数对所有用户和会话都可用是生产环境的标准做法。4.4 实操现场记录一个复杂的JSON解析UDF让我们看一个更贴近生产的例子解析用户行为日志中的JSON字段。日志中有一个extra_info字段是JSON字符串我们需要从中提取device_model和app_version。import org.apache.hadoop.hive.ql.exec.UDF; import org.apache.hadoop.io.Text; import org.json.JSONObject; // 可以使用org.json库 import org.json.JSONException; public class ParseJsonUDF extends UDF { private Text result new Text(); // 复用对象减少GC public Text evaluate(Text jsonStr, Text key) { if (jsonStr null || key null) { return null; } try { JSONObject json new JSONObject(jsonStr.toString()); if (json.has(key.toString())) { result.set(json.getString(key.toString())); return result; } else { return null; // key不存在 } } catch (JSONException e) { // 记录解析错误但不要抛出异常导致任务失败返回null // 在实际生产中这里可以增加日志输出便于排查脏数据 return null; } } }使用方式ADD JAR /path/to/json-lib.jar; -- 别忘了添加org.json库的Jar包 ADD JAR /path/to/your-udf.jar; CREATE TEMPORARY FUNCTION json_get AS com.xxx.ParseJsonUDF; SELECT user_id, json_get(extra_info, device_model) as device, json_get(extra_info, app_version) as version FROM user_log_table;这个例子展示了生产级UDF的几个要点健壮的空值和异常处理、第三方库的依赖管理、以及通过复用对象来优化性能。5. 常见问题、排查技巧与性能优化实录即使代码写对了在部署和使用过程中你依然会遇到各种各样的问题。下面是我在多年运维中积累的一些典型问题及其解决方案。5.1 常见问题速查表问题现象可能原因排查步骤与解决方案ClassNotFoundException或NoClassDefFoundError1. Jar包未正确添加到会话。2. Jar包中缺少依赖。3. Hive Server的classpath配置问题。1. 确认ADD JAR命令执行成功且路径正确HDFS路径需有权限。2. 使用mvn dependency:tree检查并打包所有非provided依赖到UDF Jar生成fat jar或使用ADD JAR依次添加所有依赖Jar。3. 联系集群管理员确认Hive Server的hive.aux.jars.path配置是否包含常用UDF路径。FAILED: SemanticException [Error 10011]: Invalid function1. 函数名重复或冲突。2. 创建函数时指定的类名错误。1. 使用SHOW FUNCTIONS LIKE *your_func*;查看是否已存在同名函数。临时函数和永久函数是分开的命名空间。2. 仔细检查CREATE FUNCTION语句中的全限定类名确保与Jar包中的类路径完全一致。UDF返回结果全是NULL1. UDF代码逻辑中未处理输入为null的情况直接返回null。2. 数据类型不匹配Hive进行了隐式转换失败。3. 业务逻辑本身导致无输出。1. 在UDF的evaluate方法开始处增加日志打印输入参数确认数据是否正常传入。2. 检查Hive表中字段类型与UDF方法声明的参数类型Text,IntWritable等是否兼容。3. 简化UDF逻辑先写一个返回固定值的版本进行测试排除业务代码问题。UDTF与LATERAL VIEW联用时报错或结果不对1. UDTF输出的列数与LATERAL VIEW ... AS后面指定的别名数量不匹配。2. UDTF的forward方法输出的对象类型与initialize声明的类型不一致。1. 确认initialize方法中定义的输出列数量fieldNames的size与AS后的别名数量一致。2. 在forward方法中打断点或打印日志确认每次输出的Object[]数组长度和内容是否符合预期。UDAF在分布式运行时结果错误1.merge方法逻辑错误合并状态时出错。2. 中间结果terminatePartial返回值序列化/反序列化有问题。3. 聚合缓冲区AggregationBuffer的reset方法未正确初始化。1.这是最棘手的。首先在本地模式下set hive.exec.mode.local.autotrue;测试小数据集结果正确后再测分布式。2. 确保terminatePartial返回的对象能被对应的ObjectInspector正确解析。对于复杂对象考虑使用Hive可序列化的标准类型如ArrayListDoubleWritable。3. 在getNewAggregationBuffer和reset方法中确保所有状态变量都被初始化。性能极差任务运行缓慢1. UDF/UDTF/UDAF内部有耗资源操作如频繁创建大对象、正则表达式编译、网络IO。2. 数据倾斜某些键Key对应的数据量巨大。1.Profile你的代码。避免在evaluate、process、iterate等方法内做重复初始化如Pattern.compile应放在类初始化阶段。强烈复用对象。2. 对于UDAF检查是否因某个分组数据量过大导致单个Reducer卡住。尝试通过set hive.groupby.skewindatatrue;开启倾斜优化或对数据先进行预处理。5.2 性能优化独家心得对象复用是黄金法则在UDF的evaluate方法中声明一个成员变量private Text result new Text();然后在方法内result.set(...); return result;。这能减少海量调用中产生的垃圾对象对性能提升立竿见影。对于UDTF和UDAF也要注意在forward或返回结果时尽量复用对象数组。谨慎使用复杂第三方库像org.json这样的库虽然方便但可能比较重。如果只是解析简单的JSON路径可以考虑使用更轻量级的库如Jackson或Gson甚至自己写简单的字符串解析。务必在打包时处理好依赖。利用Hive参数进行调试set hive.udtf.auto.progressfalse;可以关闭UDTF的进度报告有时能解决一些进度卡住的问题。set hive.exec.paralleltrue;开启任务并行对于多个UDF/UDTF阶段的任务有加速效果。对于UDAF可以通过set hive.map.aggrtrue;默认开启在Map端进行聚合减少Shuffle数据量。永久函数的管理生产环境建议建立规范的UDF管理流程。例如将所有的UDF Jar包统一上传到HDFS的特定目录如/data/udf_libs/并使用统一的命名规范。创建函数的SQL脚本纳入版本管理如Git。当UDF更新时需要先DROP FUNCTION再ADD JAR新版本最后CREATE FUNCTION。注意这可能会影响正在运行或依赖该函数的作业最好在业务低峰期操作。5.3 调试技巧如何看到UDF内部的日志这是新手最常问的问题。UDF运行在分布式集群的YARN容器里如何打印和查看日志使用System.err.println这是最直接的方法。在UDF代码中打印的信息会输出到该任务容器的标准错误stderr日志中。查看YARN日志首先在Hive CLI或Beeline中找到你的应用IDapplication_xxx_xxxx。通过YARN ResourceManager的Web UI通常8088端口找到该应用。点击应用进入“ApplicationMaster”的日志或者直接查看各个Map/Reduce Task的“Container Logs”。在Container日志里找到stderr文件你就能看到System.err.println输出的内容了。集成SLF4J日志框架对于更复杂的日志管理可以在UDF项目中引入slf4j-api和log4j等依赖并配置日志文件。但要注意日志文件会写在容器本地任务结束后会被清理需要配置日志聚合到HDFS才能长期查看。最后我个人最深刻的体会是自定义函数是Hive能力的延伸但它也是一把双刃剑。滥用UDF特别是低效的UDF会严重拖慢整个集群的任务速度。在决定自己写UDF之前务必先查一查Hive的内置函数是否已经能满足需求。如果非要写一定要把性能、健壮性空值、异常处理和可维护性清晰的命名和注释放在首位。将通用的、稳定的UDF固化下来形成团队的函数库能极大提升数据开发的效率和质量。