> **摘要**：本文从 UDF/UDAF/UDTF 三大函数类型、两种注册方式、弱类型 vs 强类型 UDAF、Aggregator 生命周期、性能陷阱五个维度，配合 2 张架构图 + 完整代码，彻底掌握 SparkSQL 自定义函数实现。
>
> **关键词**：UDF, UDAF, UDTF, Aggregator, functions.udf, spark.udf.register

---

## 一、三大函数类型

```
UDF  一对一: 1 行 → 1 行   (name → UPPER(name))
UDAF 多对一: N 行 → 1 行   (多行 → SUM/AVG)
UDTF 一对多: 1 行 → N 行   (一行 → explode 多行)
```

---

## 二、函数分类 & 注册

![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-2482016/22468951b620191c0772c180f4f67299.png)

### UDF 两种注册方式

```scala
import org.apache.spark.sql.functions._

// SQL 注册
spark.udf.register("myUpper", (s: String) => s.toUpperCase)
spark.sql("SELECT myUpper(name) FROM users")

// DSL 注册
val myUpperUdf = udf((s: String) => s.toUpperCase)
df.withColumn("upper_name", myUpperUdf(col("name")))
```

---

## 三、UDAF 深度对比 & Aggregator 生命周期

![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-2482016/52783d5b005445afd6ab375d314ebdcb.png)

### 强类型 Aggregator（推荐）

```scala
import org.apache.spark.sql.expressions.Aggregator

case class Average(var sum: Double, var count: Long)

object AverageAggregator extends Aggregator[Double, Average, Double] {
  def zero: Average = Average(0.0, 0L)                    // 初始缓冲
  def reduce(b: Average, a: Double): Average = { b.sum += a; b.count += 1; b } // 分区内
  def merge(b1: Average, b2: Average): Average = { b1.sum += b2.sum; b1.count += b2.count; b1 } // 跨分区
  def finish(reduction: Average): Double = reduction.sum / reduction.count // 输出
  def bufferEncoder: Encoder[Average] = Encoders.product
  def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

val avgUDAF = AverageAggregator.toColumn.name("avg_score")
ds.select(avgUDAF).show()
```

### 弱类型 UserDefinedAggregateFunction

```scala
class MyAvgUDAF extends UserDefinedAggregateFunction {
  def inputSchema = StructType(StructField("input", DoubleType) :: Nil)
  def bufferSchema = StructType(StructField("sum", DoubleType) :: StructField("count", LongType) :: Nil)
  def dataType = DoubleType
  def deterministic = true
  def initialize(buffer: MutableAggregationBuffer) = { buffer(0) = 0.0; buffer(1) = 0L }
  def update(buffer: MutableAggregationBuffer, input: Row) = { /* 累加 */ }
  def merge(b1: MutableAggregationBuffer, b2: Row) = { /* 合并 */ }
  def evaluate(buffer: Row) = buffer.getDouble(0) / buffer.getLong(1)
}
spark.udf.register("myAvg", new MyAvgUDAF)
```

---

## 四、性能陷阱与最佳实践

```
⚠️ UDF 是黑盒 → Catalyst 无法优化
  · 无法谓词下推 · 无法 WholeStageCodegen · 逐行序列化调用

✅ 优化建议:
  1. 优先用 Spark SQL 内置函数
  2. 复杂逻辑用 Scala 表达式组合
  3. 必须用 UDF → Pandas UDF (Arrow 向量化, 快 100x)
  4. UDAF 优先用强类型 Aggregator
```

---

## 五、总结

- **分类**：UDF 一对一 / UDAF 多对一 / UDTF 一对多
- **注册**：SQL 用 register，DSL 用 functions.udf
- **UDAF**：强类型 Aggregator 优于弱类型 UDAF

---

> **作者**：大数据技术实践者  
> **博客**：[blog.starzy.cn](https://blog.starzy.cn)  
> **GitHub**：[starzy1990.github.io](https://starzy1990.github.io)  
> **专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践**

摘要：本文从 UDF/UDAF/UDTF 三大函数类型、两种注册方式、弱类型 vs 强类型 UDAF、Aggregator 生命周期、性能陷阱五个维度，配合 2 张架构图 + 完整代码，彻底掌握 SparkSQL 自定义函数实现。

SparkSQL 之 UDF、UDAF 函数代码实现

大数据

数据库

编程语言

从UDF/UDAF/UDTF三大函数类型到双注册方式，深度解析SparkSQL自定义函数实现。对比强类型Aggregator与弱类型UDAF生命周期，揭示性能优化关键，含2张架构图及完整代码，助您规避UDF黑盒陷阱，实现高效数据处理。

数据架构

代码解释

LangGraph

GitHub

Spark 

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

SparkSQL 之 UDF、UDAF 函数代码实现-腾讯云开发者社区-腾讯云

SparkSQL 之 UDF、UDAF 函数代码实现

SparkSQL 之 UDF、UDAF 函数代码实现

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐