第15课：PySpark｜自定义函数UDF、UDAF、UDTF开发原理与企业实战

文章目录

• • 一、课前导读
  • 二、学习目标
  • 三、核心理论知识点
  • 四、原理通俗讲解
  • • 4.1 传统UDF的“逐行之旅”
    • 4.2 Pandas UDF的“批量加速”
    • 4.3 UDAF的自定义聚合
    • 4.4 UDTF的表生成
  • 五、重点概念拆解
  • • 5.1 传统UDF的注册与使用
    • 5.2 Pandas UDF的类型
    • 5.3 分组聚合Pandas UDF
    • 5.4 分组变换（GROUPED_MAP）
    • 5.5 迭代器Pandas UDF（处理大数据分区）
    • 5.6 注册UDF供SQL使用
    • 5.7 UDAF的传统实现（Scala互操作）
    • 5.8 UDTF的模拟
  • 六、易错点避坑
  • • 6.1 UDF中捕获不可序列化对象
    • 6.2 误以为UDF可以修改全局变量
    • 6.3 忽略空值导致UDF抛异常
    • 6.4 Pandas UDF返回类型不匹配
    • 6.5 分组聚合Pandas UDF中误用`agg`
    • 6.6 使用Pandas UDF时不设置`spark.sql.execution.arrow.pyspark.enabled`
  • 七、完整实战案例
  • 八、代码逐行解析
  • • 8.1 传统UDF定义
    • 8.2 Pandas UDF类型提示风格
    • 8.3 分组聚合（GROUPED_MAP）
    • 8.4 UDTF模拟：`mapInPandas`
    • 8.5 迭代器UDF（SCALAR_ITER）
  • 九、业务场景落地应用
  • • 9.1 场景一：实时日志中的IP地理位置解析
    • 9.2 场景二：用户行为序列的特征工程
    • 9.3 场景三：调用外部预测服务
    • 9.4 场景四：自定义聚合窗口函数
  • 十、常见报错排查
  • • 10.1 `AttributeError: 'NoneType' object has no attribute…`
    • 10.2 `ArrowTypeError: Expected a string, got …`
    • 10.3 `Py4JError: An error occurred while calling o123.__getstate__`
    • 10.4 Pandas UDF性能差（与内置函数差距大）
    • 10.5 分组聚合UDF内存溢出
  • 十一、本节课知识点总结
  • • UDF类型对比
    • 选型建议
    • 性能优化参数
  • 十二、课后思考作业
  • • 作业一：理论理解题
    • 作业二：代码实践题
    • 作业三：场景应用题
    • 作业四：拓展研究
• 🔗《20节课 PySpark 从入门到精通》系列课程导航

一、课前导读

在前面的课程中，我们反复强调：优先使用内置函数，避免自定义函数。因为内置函数经过Catalyst优化和代码生成，性能远超普通的Python UDF。然而，在实际的大数据开发中，总会遇到内置函数无法满足的业务逻辑——复杂的加密解密、调用外部API、非标准的业务计算公式、自定义聚合逻辑等。这些场景下，我们必须使用自定义函数。

PySpark提供了三种自定义函数接口：

• UDF（User-Defined Function）：输入一行，输出一列，最常用
• UDAF（User-Defined Aggregate Function）：自定义聚合函数，用于groupBy中复杂的状态聚合
• UDTF（User-Defined Table Function）：输入一行，输出多行（表），PySpark中通过explode或pandas UDF模拟

很多开发者在编写UDF时，由于不了解其内部原理，写出的代码效率低下——逐行处理、频繁序列化、忽略类型提示、未处理空值等。更严重的是，某些UDF写法会导致数据倾斜或内存溢出。同时，PySpark 2.3+引入了Pandas UDF（向量化UDF），利用Pandas的批量处理大幅提升了Python UDF的性能，但很多开发者仍然在用传统的逐行UDF。

本节课将全面讲解PySpark自定义函数的开发原理和最佳实践。内容包括：

• 传统UDF的注册、使用、性能陷阱及优化
• Pandas UDF（Series to Series、Series to Scalar、Iterator等）的用法
• 分组聚合UDAF的实现（通过Pandas UDF实现高效聚合）
• UDTF的模拟实现
• 企业级实战案例：如复杂JSON解析、AES加解密、IP地址解析、自定义评分模型等
• 性能对比和选型建议

学完这节课，你将能够在必须自定义逻辑的场景下，选择最高效的方式编写UDF，避免常见的性能陷阱。

二、学习目标

完成本节课的学习后，你将能够：

三、核心理论知识点

四、原理通俗讲解

4.1 传统UDF的“逐行之旅”

当你注册一个传统UDF，并应用于DataFrame时，执行过程如下：

这个过程中，每条数据都要经历多次序列化/反序列化，即使你的函数很简单，开销也很大。例如，lambda x: x+1用UDF实现可能比内置col+1慢几十倍。

4.2 Pandas UDF的“批量加速”

Pandas UDF利用Apache Arrow作为列式内存格式，批量传输数据。流程如下：

这样大大减少了JVM与Python之间的通信次数，并且利用Pandas底层C语言加速，性能通常比逐行UDF高一个数量级，接近内置函数的水平。

4.3 UDAF的自定义聚合

Spark内置聚合函数（如sum、avg）无法满足复杂聚合需求时，可以自定义聚合。在PySpark中，推荐使用grouped_apply或applyInPandas配合Pandas UDF实现。你定义一个函数，接收一个分组的所有数据（Pandas DataFrame），返回一个聚合结果（Pandas Series或DataFrame）。Spark会自动将分组数据拉取到同一个Executor上（可能引发shuffle），然后调用你的函数。

4.4 UDTF的表生成

Spark SQL原生不支持UDTF（类似Hive的explode），但可以通过flatMap或explode函数模拟。例如，你可以定义UDF返回一个数组，然后调用explode将数组展开为多行，效果等同于UDTF。Pandas UDF也可以返回pd.DataFrame来实现多行输出。

五、重点概念拆解

5.1 传统UDF的注册与使用

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# 方式1：装饰器
@udf(returnType=StringType())
def my_upper(s):
if s is None:
return None
return s.upper()

# 方式2：函数定义后注册
def my_lower(s):
return s.lower() if s else None
my_lower_udf = udf(my_lower, StringType())

# 使用
df.withColumn("upper_name", my_upper(col("name")))

关键点：

• 必须指定返回类型（或Spark自动推断，但建议显式指定）。
• 函数内部应处理空值（None），否则可能抛异常。
• UDF默认是确定性的（相同输入相同输出），可通过udf(…, deterministic=False)声明非确定性（如随机函数）。

5.2 Pandas UDF的类型

Pandas UDF通过@pandas_udf装饰器，并传入返回类型和函数类型。函数类型有：

重要：PySpark 3.0+推荐使用类型提示的方式定义Pandas UDF，而不使用PandasUDFType枚举。例如：

from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf(StringType())
def my_upper(s: pd.Series) –> pd.Series:
return s.str.upper()

5.3 分组聚合Pandas UDF

示例：自定义聚合函数——计算每个组的几何平均值。

@pandas_udf(returnType=FloatType())
def geometric_mean(v: pd.Series) –> float:
return np.exp(np.log(v).mean())

df.groupBy("group").agg(geometric_mean(col("value")).alias("geo_mean"))

GROUPED_AGG类型的Pandas UDF接收每个分组的列（一个或多个Series），返回一个标量值。

5.4 分组变换（GROUPED_MAP）

@pandas_udf(returnType=df.schema, functionType=PandasUDFType.GROUPED_MAP)
def normalize(df):
df["value_norm"] = (df["value"] – df["value"].mean()) / df["value"].std()
return df

df.groupBy("group").apply(normalize)

注意：GROUPED_MAP要求返回的DataFrame必须与输入的StructType兼容，但行数可以不同。

5.5 迭代器Pandas UDF（处理大数据分区）

当分区内数据量大无法一次性装入内存时，可以使用SCALAR_ITER或MAP_ITER。例如：

@pandas_udf(StringType())
def process_batch(iterator: Iterator[pd.Series]) –> Iterator[pd.Series]:
for s in iterator:
yield s.str.upper()

Spark会将每个分区的数据分批次传入迭代器，每次处理一个批次（由spark.sql.execution.arrow.maxRecordsPerBatch控制）。

5.6 注册UDF供SQL使用

spark.udf.register("my_upper", lambda s: s.upper() if s else None, StringType())
spark.sql("SELECT my_upper(name) FROM users").show()

5.7 UDAF的传统实现（Scala互操作）

在PySpark中，如果你需要高性能的UDAF，可以编写Scala UDAF并通过spark.udf.registerJavaFunction调用。但一般不推荐，因为增加了复杂度。Pandas UDF的GROUPED_AGG已经能满足大多数需求。

5.8 UDTF的模拟

方法1：UDF返回数组 + explode

def split_words(s):
return s.split() if s else []
split_udf = udf(split_words, ArrayType(StringType()))
df.withColumn("word", explode(split_udf(col("text"))))

方法2：使用flatMap（RDD层）

df.rdd.flatMap(lambda row: [(row.id, word) for word in row.text.split()]).toDF(["id", "word"])

方法3：mapInPandas返回多行

def generate_rows(iterator):
for pdf in iterator:
for _, row in pdf.iterrows():
for word in row["text"].split():
yield (row["id"], word)

df.mapInPandas(generate_rows, schema="id int, word string")

六、易错点避坑

6.1 UDF中捕获不可序列化对象

错误示例：

class MyProcessor:
def __init__(self, data):
self.data = data
def process(self, x):
return self.data[x]

processor = MyProcessor({"a":1})
udf(processor.process, IntegerType()) # 报错：无法序列化processor

解决：将数据通过广播变量传递，或者在UDF内部重新创建对象（如果轻量）。

6.2 误以为UDF可以修改全局变量

UDF在Executor的Python进程中执行，修改的全局变量不会影响Driver或其他Executor。如需累加，使用累加器。

6.3 忽略空值导致UDF抛异常

UDF接收的输入可能为None，必须显式处理，否则抛出TypeError或AttributeError。

6.4 Pandas UDF返回类型不匹配

错误：声明返回FloatType()，但实际返回了pd.Series包含字符串。

解决：确保返回的Series类型与声明的Spark类型兼容（如float64对应DoubleType）。

6.5 分组聚合Pandas UDF中误用agg

# 错误：不能直接在agg中使用GROUPED_MAP类型的UDF
df.groupBy("g").agg(my_udf(df.value)) # 报错

GROUPED_MAP通过apply使用，GROUPED_AGG通过agg使用。

6.6 使用Pandas UDF时不设置spark.sql.execution.arrow.pyspark.enabled

Pandas UDF依赖Arrow，该配置默认true（Spark 3.x），但如果设置为false，Pandas UDF会回退到逐行处理，性能下降严重。

七、完整实战案例

本案例将演示从传统UDF到Pandas UDF的演进，并实现一个复杂的业务场景：IP地址解析（IP转换为地理位置）、自定义聚合（计算每个用户会话的停留时长）、UDTF模拟（URL拆解为路径层级）。

# ============== udf_udaf_udtf_demo.py ==============
# 功能：PySpark自定义函数（UDF、Pandas UDF、UDAF、UDTF模拟）企业级实战
# 数据：模拟用户访问日志（IP、时间、URL）
# 环境：PySpark 3.x，需要安装geoip2库（可选，这里模拟）

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, pandas_udf, col, explode, struct, lit
from pyspark.sql.types import StringType, IntegerType, DoubleType, ArrayType, StructType, StructField, TimestampType
import pandas as pd
import numpy as np
import random
from datetime import datetime, timedelta
import time

# ========== 1. 创建SparkSession ==========
spark = SparkSession.builder \\
.appName("UDF_UDAF_UDTF_Demo") \\
.master("local[4]") \\
.config("spark.sql.execution.arrow.pyspark.enabled", "true") \\
.config("spark.sql.execution.arrow.maxRecordsPerBatch", "10000") \\
.getOrCreate()

sc = spark.sparkContext
sc.setLogLevel("WARN")

print("=" * 80)
print("PySpark 自定义函数（UDF / Pandas UDF / UDAF / UDTF）企业实战")
print("=" * 80)

# ========== 2. 生成模拟数据 ==========
print("\\n生成用户访问日志…")
num_records = 500_000
users = [f"user_{i}" for i in range(1, 1001)]
ips = [f"192.168.{random.randint(1,255)}.{random.randint(1,255)}" for _ in range(500)]
urls = ["/home", "/product/1", "/product/2", "/cart", "/checkout", "/api/status", "/search?q=spark"]

def generate_log(i):
user = random.choice(users)
ip = random.choice(ips)
url = random.choice(urls)
ts = datetime.now() – timedelta(seconds=random.randint(0, 86400 * 30))
return (user, ip, url, ts)

data = [generate_log(i) for i in range(num_records)]
df = spark.createDataFrame(data, ["user_id", "ip", "url", "event_time"])
print(f"生成了 {df.count():,} 条日志")
df.show(5, truncate=False)

# ========== 3. 传统UDF示例：IP地址段解析（模拟地理位置） ==========
print("\\n" + "=" * 80)
print("步骤1: 传统UDF – IP地址解析")
print("=" * 80)

# 模拟IP到省份的映射（实际可用geoip数据库）
ip_to_city = {f"192.168.{i}.{j}": f"City_{i % 5}" for i in range(1, 256) for j in range(1, 256)}
# 简化：只模拟部分规则
def ip_to_region(ip):
if not ip:
return "Unknown"
parts = ip.split(".")
if len(parts) == 4:
second = int(parts[1])
if second <= 50:
return "North"
elif second <= 100:
return "East"
elif second <= 150:
return "South"
else:
return "West"
return "Unknown"

# 注册传统UDF
region_udf = udf(ip_to_region, StringType())
df_with_region = df.withColumn("region", region_udf(col("ip")))
print("添加地理位置列（传统UDF）:")
df_with_region.select("ip", "region").show(10, truncate=False)

# 性能测试：传统UDF
start = time.time()
df_with_region.count()
print(f"传统UDF处理500k条耗时: {time.time() – start:.2f}秒")

# ========== 4. Pandas UDF（Series to Series）替代传统UDF ==========
print("\\n" + "=" * 80)
print("步骤2: Pandas UDF向量化处理")
print("=" * 80)

# 使用类型提示的Pandas UDF（推荐）
@pandas_udf(StringType())
def ip_to_region_pandas(ip_series: pd.Series) –> pd.Series:
# 向量化操作：使用str访问器和条件判断
# 提取第二段
parts = ip_series.str.split('.', expand=True)
second = parts[1].astype(int)
result = pd.Series(index=ip_series.index, dtype="object")
result[second <= 50] = "North"
result[(second > 50) & (second <= 100)] = "East"
result[(second > 100) & (second <= 150)] = "South"
result[second > 150] = "West"
result[second.isna()] = "Unknown"
return result

df_with_region_pd = df.withColumn("region", ip_to_region_pandas(col("ip")))
print("Pandas UDF结果示例:")
df_with_region_pd.select("ip", "region").show(10, truncate=False)

start = time.time()
df_with_region_pd.count()
print(f"Pandas UDF处理耗时: {time.time() – start:.2f}秒")

# ========== 5. 分组聚合UDAF：计算每个用户的会话时长 ==========
print("\\n" + "=" * 80)
print("步骤3: 自定义聚合函数（UDAF）- 用户会话分析")
print("=" * 80)

# 会话定义：同一用户相邻事件间隔超过30分钟视为新会话
# 我们需要计算每个用户的总会话时长（所有会话时长之和）
# 使用Pandas UDF GROUPED_AGG实现自定义聚合

# 先按用户和时间排序，分组数据传入一个DataFrame
def session_duration_udaf(pdf: pd.DataFrame) –> float:
# 假设pdf有event_time列，已按时间排序
# 计算会话间隙，超过30分钟切分
pdf = pdf.sort_values("event_time")
deltas = pdf["event_time"].diff().dt.total_seconds().fillna(0)
# 标记新会话开始（间隙>1800秒）
session_starts = deltas > 1800
session_id = session_starts.cumsum()
# 计算每个会话的时长（最大值-最小值）
durations = pdf.groupby(session_id)["event_time"].agg(lambda x: x.max() – x.min()).dt.total_seconds()
total_duration = durations.sum()
return total_duration

# 注册为聚合UDF（GROUPED_AGG）
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType

session_udaf = pandas_udf(session_duration_udaf, returnType=DoubleType())

# 注意：GROUPED_AGG类型只能用于聚合，输入是一组列（此处我们要传入完整DataFrame结构）
# 更常用的是对分组后的每个Group应用自定义聚合函数，需要传递多个列作为参数。
# 但GROUPED_AGG UDF参数必须是多个Series，不能直接接收DataFrame。
# 因此我们用另一种方式：groupBy().apply() 的GROUPED_MAP类型。

# 方法：使用GROUPED_MAP，返回聚合结果
@pandas_udf(returnType=StructType([StructField("total_session_duration", DoubleType())]),
functionType="GROUPED_MAP")
def session_analysis(pdf: pd.DataFrame) –> pd.DataFrame:
pdf = pdf.sort_values("event_time")
deltas = pdf["event_time"].diff().dt.total_seconds().fillna(0)
session_starts = deltas > 1800
session_id = session_starts.cumsum()
durations = pdf.groupby(session_id)["event_time"].agg(lambda x: x.max() – x.min()).dt.total_seconds()
total_duration = durations.sum()
return pd.DataFrame({"total_session_duration": [total_duration]})

# 应用
result = df.groupBy("user_id").apply(session_analysis)
print("每个用户的总会话时长（秒）:")
result.orderBy(col("total_session_duration").desc()).show(10)

# 更简洁的UDAF方式：使用aggregate + Pandas UDF（需要Spark 3.0+）
# 这里我们展示另一种：使用内置函数配合窗口，但为了演示UDAF，采用GROUPED_MAP。

# ========== 6. 模拟UDTF：将URL拆分为路径层级 ==========
print("\\n" + "=" * 80)
print("步骤4: UDTF模拟 – URL路径拆分")
print("=" * 80)

# 传统方式：UDF返回数组 + explode
def url_to_paths(url: str):
if not url:
return []
# 去掉查询参数
path = url.split('?')[0]
parts = path.split('/')
# 过滤空字符串
return [p for p in parts if p]

paths_udf = udf(url_to_paths, ArrayType(StringType()))
df_exploded = df.withColumn("path_parts", paths_udf(col("url"))) \\
.withColumn("path_level", explode(col("path_parts")))
print("URL拆解后的路径组件:")
df_exploded.select("url", "path_level").show(10, truncate=False)

# 使用Pandas UDF + mapInPandas 实现更复杂的UDTF（可以输出多行多列）
def url_to_rows(iterator: Iterator[pd.DataFrame]) –> Iterator[pd.DataFrame]:
for pdf in iterator:
# 对每个DataFrame（分区）逐行处理
rows = []
for _, row in pdf.iterrows():
url = row["url"]
user_id = row["user_id"]
if url:
path = url.split('?')[0]
for level, part in enumerate(path.split('/')):
if part:
rows.append({"user_id": user_id, "url": url, "level": level, "part": part})
yield pd.DataFrame(rows)

schema = StructType([
StructField("user_id", StringType()),
StructField("url", StringType()),
StructField("level", IntegerType()),
StructField("part", StringType())
])

df_udtf = df.select("user_id", "url").mapInPandas(url_to_rows, schema=schema)
print("mapInPandas UDTF结果:")
df_udtf.show(10, truncate=False)

# ========== 7. 复杂Pandas UDF：调用外部模型（模拟） ==========
print("\\n" + "=" * 80)
print("步骤5: 复杂Pandas UDF – 评分模型预测")
print("=" * 80)

# 模拟一个预先训练好的模型（简单线性回归系数）
# 假设我们有一个特征向量 [浏览时长, 点击次数, 页面深度] 预测用户评分
# 特征数据由url长度、点击数等模拟

# 生成模拟特征数据
feature_df = df.withColumn("url_length", length(col("url"))) \\
.withColumn("hour", hour(col("event_time"))) \\
.withColumn("is_api", col("url").contains("api"))

# 定义模型预测函数（向量化）
coef = np.array([0.1, –0.05, 0.3]) # 权重

@pandas_udf(DoubleType())
def predict_score(url_length: pd.Series, hour: pd.Series, is_api: pd.Series) –> pd.Series:
# 构建特征矩阵
features = pd.DataFrame({
"len": url_length,
"hour": hour,
"is_api": is_api.astype(int)
}).values
# 预测
scores = features.dot(coef) + 0.5 # 加偏置
# 限制范围[0,1]
return pd.Series(np.clip(scores, 0, 1))

df_with_score = feature_df.withColumn("score", predict_score(col("url_length"), col("hour"), col("is_api")))
print("评分结果示例:")
df_with_score.select("url", "url_length", "hour", "is_api", "score").show(10)

# ========== 8. 迭代器Pandas UDF处理超大分区 ==========
print("\\n" + "=" * 80)
print("步骤6: 迭代器Pandas UDF（分批次处理）")
print("=" * 80)

# 场景：需要逐行调用外部API，但内存有限，分批次发送
def call_external_api(iterator: Iterator[pd.Series]) –> Iterator[pd.Series]:
for batch in iterator:
# batch是pd.Series，假设是url列
results = []
for url in batch:
# 模拟API调用
result = len(url) # 简单模拟
results.append(result)
yield pd.Series(results)

# 注册迭代器UDF
@pandas_udf(IntegerType(), functionType="SCALAR_ITER")
def api_udf(iterator):
for batch in iterator:
yield batch.str.len() # 向量化计算长度，演示

# 或者使用mapInPandas
def api_map(iterator):
for pdf in iterator:
pdf["url_len"] = pdf["url"].str.len()
yield pdf[["user_id", "url_len"]]

df_len = df.select("user_id", "url").mapInPandas(api_map, schema="user_id string, url_len int")
df_len.show(5)

# ========== 9. 注册UDF供SQL使用 ==========
print("\\n" + "=" * 80)
print("步骤7: 注册UDF到SQL引擎")
print("=" * 80)

# 注册传统UDF
spark.udf.register("region_udf_sql", ip_to_region, StringType())
# 注册Pandas UDF到SQL（Spark 3.0+支持）
spark.udf.register("region_pandas_sql", ip_to_region_pandas, StringType())

df.createOrReplaceTempView("logs")
sql_result = spark.sql("""
SELECT ip, region_udf_sql(ip) as region_traditional, region_pandas_sql(ip) as region_pandas
FROM logs
LIMIT 10
""")
print("SQL中使用UDF:")
sql_result.show(truncate=False)

# ========== 10. 性能对比汇总 ==========
print("\\n" + "=" * 80)
print("步骤8: 性能对比总结")
print("=" * 80)

# 重新计时对比
def benchmark(udf_type):
if udf_type == "traditional":
df.withColumn("region", region_udf(col("ip"))).count()
elif udf_type == "pandas":
df.withColumn("region", ip_to_region_pandas(col("ip"))).count()
elif udf_type == "builtin":
# 模拟内置函数替代（实际上不能用内置函数直接映射IP，仅示意）
df.withColumn("region", lit("East")).count()

for t in ["traditional", "pandas"]:
start = time.time()
benchmark(t)
elapsed = time.time() – start
print(f"{t} UDF 耗时: {elapsed:.2f}秒")

print("\\n性能总结: Pandas UDF（向量化）比传统UDF快3-10倍，接近内置函数。")

# ========== 11. 清理 ==========
spark.stop()
print("\\n✅ 自定义函数实战演示完成")

八、代码逐行解析

8.1 传统UDF定义

@udf(returnType=StringType())
def ip_to_region(ip):
if not ip:
return "Unknown"
# 逻辑…

• 装饰器@udf需要指定返回类型（否则Spark尝试推断，可能出错）。
• 函数内部必须处理空值。

8.2 Pandas UDF类型提示风格

@pandas_udf(StringType())
def ip_to_region_pandas(ip_series: pd.Series) –> pd.Series:
return ip_series.str.split('.').str[1].astype(int).apply(lambda x: ...)

• 使用Python类型提示更加清晰。
• 利用Pandas的字符串向量化方法（.str）提高性能。

8.3 分组聚合（GROUPED_MAP）

@pandas_udf(returnType=StructType([...]), functionType="GROUPED_MAP")
def session_analysis(pdf: pd.DataFrame) –> pd.DataFrame:
# 处理一组数据
return pd.DataFrame({"col": [value]})

• 接受一个分组的全部数据作为Pandas DataFrame，输出一个DataFrame（可以是多行）。
• 与groupBy().apply()配合使用，每个分组调用一次。

8.4 UDTF模拟：mapInPandas

def url_to_rows(iterator: Iterator[pd.DataFrame]) –> Iterator[pd.DataFrame]:
for pdf in iterator:
# 处理并yield新的DataFrame
yield new_pdf
df.mapInPandas(url_to_rows, schema)

• 更灵活，可以输出任意行数和列数。

8.5 迭代器UDF（SCALAR_ITER）

@pandas_udf(IntegerType(), functionType="SCALAR_ITER")
def api_udf(iterator):
for batch in iterator:
yield batch.str.len()

• 每个批次一个pd.Series，避免一次性加载整个分区。

九、业务场景落地应用

9.1 场景一：实时日志中的IP地理位置解析

实际生产使用GeoIP数据库（如MaxMind），通过Pandas UDF批量查询内存中的IP索引，大幅提升吞吐量。

9.2 场景二：用户行为序列的特征工程

对于每个用户的事件序列，需要计算序列统计特征（如平均间隔、序列熵等）。使用groupBy().applyInPandas，将每个用户的数据加载到Pandas DataFrame，利用Python的统计库计算复杂特征。

9.3 场景三：调用外部预测服务

在流式计算中，每条数据需要调用外部模型服务（HTTP）。可以使用mapInPandas批量聚合请求，减少网络开销。

9.4 场景四：自定义聚合窗口函数

Spark内置窗口函数不支持自定义复杂聚合（如中位数、模式），可以使用Pandas UDF配合groupBy实现。

十、常见报错排查

10.1 AttributeError: 'NoneType' object has no attribute…

原因：UDF中没有处理输入为None的情况。

解决：在函数开始检查if s is None: return None。

10.2 ArrowTypeError: Expected a string, got …

原因：Pandas UDF返回的Series类型与声明的Spark类型不匹配。

解决：确保返回的Series dtype对应正确，如返回pd.Series的dtype为object或string，与StringType兼容。

10.3 Py4JError: An error occurred while calling o123.__getstate__

原因：UDF中捕获了不可序列化的对象（如网络连接）。

解决：在函数内部创建连接，或使用广播变量。

10.4 Pandas UDF性能差（与内置函数差距大）

原因：Arrow未启用，或数据量太小批次效果不明显，或Pandas操作非向量化。

解决：确认spark.sql.execution.arrow.pyspark.enabled=true，且Pandas操作使用向量化方法（如.str、.dt），避免apply循环。

10.5 分组聚合UDF内存溢出

原因：单个分组的数据量过大，装入Pandas DataFrame导致OOM。

解决：对分组内数据采样或限制大小；使用mapInPandas逐步处理。

十一、本节课知识点总结

UDF类型对比

选型建议

• 能用内置函数绝不用UDF
• 必须用UDF时，优先选择Pandas UDF
• 如果数据量大且分组均匀，可用GROUPED_MAP
• 如果输出多行，使用mapInPandas或explode+UDF

性能优化参数

十二、课后思考作业

作业一：理论理解题

作业二：代码实践题

作业三：场景应用题

某风控系统需要实时处理交易事件，每条交易包含user_id、amount、timestamp、location。需要：

• 对于每个用户，计算过去1小时内的交易总金额（滑动窗口聚合）
• 标记交易金额超过该用户历史平均金额3倍的交易（异常检测）
• 调用外部规则引擎（HTTP接口）获取风险评分，评分结果需要更新回DataFrame

请设计使用PySpark UDF/Pandas UDF的解决方案，说明选择的UDF类型及原因，并写出关键代码。

作业四：拓展研究

提交方式：本次作业要求提交可运行的代码和性能对比数据，以及理论题的文字解答。鼓励将UDF封装为可导入的模块，提高复用性。

扩展阅读：

• Spark官方文档：Pandas UDFs
• 《PySpark实战指南》第8章
• Apache Arrow官方文档

通过本节课的学习，你已经掌握了PySpark自定义函数的全套技能，从传统UDF到高性能Pandas UDF，再到UDAF和UDTF模拟。记住：首选内置函数，不得已用UDF时首选Pandas UDF。下一节课我们将学习PySpark多数据源读写（CSV/JSON/Parquet/Hive/MySQL），让Spark与外部存储系统无缝集成。我们下节课见！

🔗《20节课 PySpark 从入门到精通》系列课程导航

去订阅

🌟 感谢您耐心阅读到这里！ 💡 如果本文对您有所启发欢迎： 👍 点赞📌 收藏 📤 分享给更多需要的伙伴。 🗣️ 期待在评论区看到您的想法, 共同进步。 🔔 关注我，持续获取更多干货内容～ 🤗 我们下篇文章见～