# 一、概念
## UDF（User Defined Function，用户自定义函数）
UDF 的核心逻辑是**一行进、一个值出**，本质上是对表中每一行的某个字段做一次独立的转换或计算，结果作为新的一列返回。你可以把它理解为对每一行数据套用同一个"加工模板"。

## UDAF（User Defined Aggregate Function，用户自定义聚合函数）
UDAF 的核心逻辑是**多行进、一个值出**，它需要对一组数据进行汇总计算，最终输出一个聚合结果。内置的 SUM、AVG、COUNT 都是这个思路，UDAF 允许你自定义聚合的计算规则。

## UDTF（User Defined Table Function，用户自定义表函数）
UDTF 的核心逻辑是**一行进、多行出**，它能把一条记录"炸开"成多条记录，输出的是一张"小表"。这是三者中输出形态最特殊的一种，内置的 explode() 函数就是这个思想的体现。

# 二、技术学习详情
## 2.1 环境搭建
完成本地 PySpark 开发环境配置，技术栈如下：

| 组件 | 版本 |
|------|------|
| Python | 3.10.19 |
| OpenJDK (Zulu) | 11.0.30 |
| PySpark | 3.5.1 |
| PyArrow | 最新稳定版 |
| Pandas | 最新稳定版 |

```bash
# 安装 PySpark 及依赖（pyarrow 是 UDAF 所需）
pip install pyspark==3.5.1 pyarrow pandas
```
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-11994342/9b2a8b4717d0abb74dd6bb24e2ecc379.png)

## 2.2 核心内容：PySpark 三类自定义函数
### UDF — 自定义标量函数
实现了一个分数等级转换函数 score_to_grade，掌握了两种调用方式：
```python
@udf(returnType=StringType())
def score_to_grade(score):
    if score is None: return "缺考"
    elif score >= 90: return "A（优秀）"
    elif score >= 75: return "B（良好）"
    elif score >= 60: return "C（及格）"
    else:             return "D（不及格）"
```

### UDAF — 自定义聚合函数
实现了去极值均值 trimmed_mean（基于 Pandas UDF），原理为去除最大值和最小值后再取平均，可有效抵抗异常值对均值的干扰：
```python
@pandas_udf("double")
def trimmed_mean(values: pd.Series) -> float:
    clean = values.dropna()
    if len(clean) <= 2:
        return float(clean.mean())
    trimmed = clean.sort_values().iloc[1:-1]   # 去掉首尾极值
    return float(trimmed.mean())
```

### UDTF — 自定义表函数
实现了句子分词器 SentenceTokenizer，将一段文本拆分为多行，每行包含词语内容、位置索引和字符数：
```python
@udtf(returnType="word: string, position: int, char_count: int")
class SentenceTokenizer:
    def eval(self, sentence: str):
        if sentence is None:
            return
        for idx, word in enumerate(sentence.split()):
            yield (word, idx, len(word))   # 一行 → 多行输出
```

## 2.3 完整程序及结果
```python
os.environ['PYSPARK_PYTHON'] = r'C:\Users\51591\.conda\envs\spark_env\python.exe'
os.environ['PYSPARK_DRIVER_PYTHON'] = r'C:\Users\51591\.conda\envs\spark_env\python.exe'

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.functions import udf, pandas_udf, udtf, col, lit
from pyspark.sql.types import StringType, DoubleType
import pandas as pd

# ── 初始化 SparkSession ──────────────────────────────────────
spark = SparkSession.builder \
    .appName("UDF_UDAF_UDTF_Demo") \
    .master("local[*]") \
    .config("spark.driver.memory", "2g") \
    .getOrCreate()

spark.sparkContext.setLogLevel("WARN")
print(f"✅ Spark 启动成功，版本: {spark.version}\n")


# ============================================================
# 一、UDF（User Defined Function）
#   作用：一行输入 → 一行输出（标量转换）
# ============================================================

@udf(returnType=StringType())
def score_to_grade(score):
    """将数字分数转换为等级标签"""
    if score is None:
        return "缺考"
    elif score >= 90: return "A（优秀）"
    elif score >= 75: return "B（良好）"
    elif score >= 60: return "C（及格）"
    else:             return "D（不及格）"

spark.udf.register("score_to_grade_sql", score_to_grade)

score_df = spark.createDataFrame([
    ("Alice",   95),
    ("Bob",     72),
    ("Charlie", 58),
    ("Diana",   88),
    ("Eve",     None),
], ["name", "score"])

print("=" * 55)
print("【UDF 测试】DataFrame API 调用")
print("=" * 55)
score_df.withColumn("grade", score_to_grade(col("score"))).show()

print("【UDF 测试】SQL 调用")
score_df.createOrReplaceTempView("students")
spark.sql("""
    SELECT name, score, score_to_grade_sql(score) AS grade
    FROM students
""").show()


# ============================================================
# 二、UDAF（User Defined Aggregate Function）
#   作用：多行输入 → 一行输出（配合 groupBy 使用）
#   注意：Pandas UDAF 不能与普通聚合函数在同一 agg() 中混用
# ============================================================

@pandas_udf("double")
def trimmed_mean(values: pd.Series) -> float:
    """
    去极值均值：去掉最高和最低值后计算平均数。
    可有效抵抗异常值（outlier）对均值的干扰。
    """
    clean = values.dropna()
    if len(clean) <= 2:
        return float(clean.mean())
    trimmed = clean.sort_values().iloc[1:-1]
    return float(trimmed.mean())

dept_df = spark.createDataFrame([
    ("研发部", "Alice",   25000.0),
    ("研发部", "Bob",     28000.0),
    ("研发部", "Charlie", 99999.0),  # 异常高薪，影响普通均值
    ("研发部", "Diana",   26500.0),
    ("市场部", "Eve",     18000.0),
    ("市场部", "Frank",   20000.0),
    ("市场部", "Grace",     100.0),  # 异常低薪，影响普通均值
    ("市场部", "Henry",   19500.0),
], ["dept", "name", "salary"])

print("=" * 55)
print("【UDAF 测试】按部门对比普通均值 vs 去极值均值")
print("=" * 55)

# ✅ 修复：Pandas UDAF 不能与普通聚合函数混用，分开计算再 join 合并
avg_df = dept_df.groupBy("dept").agg(
    F.round(F.avg("salary"), 2).alias("普通均值")
)
trimmed_df = dept_df.groupBy("dept").agg(
    F.round(trimmed_mean("salary"), 2).alias("去极值均值")
)
avg_df.join(trimmed_df, on="dept").show()


# ============================================================
# 三、UDTF（User Defined Table Function）
#   作用：一行输入 → 多行输出（表生成函数）
#   要求：PySpark >= 3.5
# ============================================================

@udtf(returnType="word: string, position: int, char_count: int")
class SentenceTokenizer:
    """
    将一个句子拆分为多行词语，
    每行附带：词语内容、在句中的位置、字符数。
    """
    def eval(self, sentence: str):
        if sentence is None:
            return
        for idx, word in enumerate(sentence.split()):
            yield (word, idx, len(word))

spark.udtf.register("sentence_tokenizer", SentenceTokenizer)

# ── 用法一：直接传入字面量调用 ──
print("=" * 55)
print("【UDTF 测试1】直接调用（字面量输入）")
print("=" * 55)
SentenceTokenizer(lit("Apache Spark PySpark 实战")).show()

# ── 用法二：SQL 直接调用 ──
print("【UDTF 测试2】SQL 直接调用")
spark.sql("SELECT * FROM sentence_tokenizer('大数据 工程师 学习 Spark')").show()

# ── 用法三：配合 DataFrame，通过 LATERAL 横向关联展开 ──
text_df = spark.createDataFrame([
    (1, "Hello Spark World"),
    (2, "UDF UDAF UDTF"),
    (3, "大数据 实战 学习"),
], ["id", "sentence"])

text_df.createOrReplaceTempView("texts")

print("【UDTF 测试3】LATERAL 横向关联 配合 DataFrame（每行展开为多行）")
# ✅ 修复：Python UDTF 不支持 Hive 的 LATERAL VIEW 语法
#         应改用 ANSI SQL 标准的 LATERAL 横向关联语法
spark.sql("""
    SELECT t.id, t.sentence, u.word, u.position, u.char_count
    FROM   texts AS t, LATERAL sentence_tokenizer(t.sentence) AS u
    ORDER BY t.id, u.position
""").show(20)

# ── 停止 Spark ──
spark.stop()
print("\n✅ 所有示例运行完成！")
```
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-11994342/6953a7dca1b57dc4b1b6ac22b05b2ea5.png)
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-11994342/f1efa2afc703f4ee649c9c4d4c4e01a9.png)
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-11994342/e9eea5edfc3e8c38bd971331024738ab.png)

# 四、问题排查与解决汇总
| 序号 | 错误信息 | 根本原因 | 解决方案 |
|------|----------|----------|----------|
| 1 | INVALID_PANDAS_UDF_PLACEMENT | Pandas UDAF 与内置聚合函数不能共存于同一 agg() | 拆分两次 groupBy 再 join |
| 2 | ROUTINE_NOT_FOUND | LATERAL VIEW 是 Hive 语法，不支持 Python UDTF | 改用 ANSI SQL LATERAL 关联语法 |
| 3 | 临时目录删除报错 | Windows 环境下 Spark 关闭时的已知问题 | 不影响运行结果，可忽略 |

UDF 的核心逻辑是一行进、一个值出，本质上是对表中每一行的某个字段做一次独立的转换或计算，结果作为新的一列返回。你可以把它理解为对每一行数据套用同一个"加工模板"。

玩转 PySpark 自定义函数：UDF、UDAF、UDTF 全解析与实战

数据开发/数据仓库

大数据

编程语言

后端

PySpark自定义函数详解：UDF实现行级转换，UDAF处理聚合计算，UDTF进行数据展开。掌握三类函数开发技巧，包括分数等级转换、去极值均值计算和文本分词处理，提升大数据处理效率。适用于PySpark 3.5+环境，提供完整代码示例和常见问题解决方案。

Windows

Spark 

Hive

2026新春采购季

tione

wedata

bigdata-class

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

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

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

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

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

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

社区新版编辑器体验调研

玩转 PySpark 自定义函数：UDF、UDAF、UDTF 全解析与实战-腾讯云开发者社区-腾讯云

玩转 PySpark 自定义函数：UDF、UDAF、UDTF 全解析与实战

玩转 PySpark 自定义函数：UDF、UDAF、UDTF 全解析与实战

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐