第14课：PySpark｜内置函数全解【字符串日期聚合条件窗口函数实战】

文章目录

• • 一、课前导读
  • 二、学习目标
  • 三、核心理论知识点
  • 四、原理通俗讲解
  • • 4.1 内置函数为何比UDF快得多？
    • 4.2 函数组合与链式调用
    • 4.3 窗口函数的物理执行
  • 五、重点概念拆解
  • • 5.1 字符串函数详解
    • 5.2 日期时间函数详解
    • 5.3 聚合函数详解
    • 5.4 条件函数详解
    • 5.5 窗口函数详解
    • 5.6 集合函数（数组与Map）
  • 六、易错点避坑
  • • 6.1 混淆`substring`的索引起始
    • 6.2 日期格式字符串大小写敏感
    • 6.3 窗口函数中`order by`缺少会导致整个分区作为一个窗口
    • 6.4 `collect_list`结果无序且可能受分区影响
    • 6.5 在聚合前使用`explode`导致数据膨胀
    • 6.6 使用`when`时遗漏`otherwise`
    • 6.7 `coalesce`与`repartition`混淆
  • 七、完整实战案例
  • 八、代码逐行解析
  • • 8.1 正则提取字段
    • 8.2 字符串分割与取元素
    • 8.3 日期转换
    • 8.4 条件分类
    • 8.5 近似去重计数
    • 8.6 透视（pivot）
    • 8.7 窗口函数
    • 8.8 数组展开
    • 8.9 UDF vs 内置函数性能对比
  • 九、业务场景落地应用
  • • 9.1 场景一：用户注册日志中的设备信息解析
    • 9.2 场景二：电商订单的RFM分层
    • 9.3 场景三：实时流处理中的JSON解析
    • 9.4 场景四：日志路径分析中的路径拆分
  • 十、常见报错排查
  • • 10.1 `to_timestamp`返回null
    • 10.2 窗口函数报错`Window function without frame`
    • 10.3 `regexp_extract`提取不到值返回空字符串
    • 10.4 使用`collect_list`导致内存溢出
    • 10.5 `explode`后数据膨胀导致Shuffle数据量巨大
  • 十一、本节课知识点总结
  • • 函数分类速查表
    • 性能优化要点
  • 十二、课后思考作业
  • • 作业一：理论理解题
    • 作业二：代码实践题
    • 作业三：场景应用题
    • 作业四：拓展研究
• 🔗《20节课 PySpark 从入门到精通》系列课程导航

一、课前导读

在前面的课程中，我们已经学习了DataFrame的基础操作和Spark SQL的完整语法。你可能已经发现，在编写DataFrame代码时，大量的数据处理逻辑都依赖于pyspark.sql.functions模块中的内置函数。例如，用when实现条件判断，用date_add处理日期，用row_number进行排名。这些内置函数不仅让代码更简洁，更重要的是——它们的执行效率远超自定义UDF，因为内置函数是直接在JVM中执行的，不需要Python和JVM之间的序列化开销。

然而，内置函数的数量极其庞大（超过200个），很多初学者只会用其中几个（如col、sum、avg），遇到稍微复杂的需求就束手无策，甚至去写低效的UDF。例如，想要提取字符串中的日期、计算两个日期的差值、实现行转列（pivot）、计算累计和等，实际上都有对应的内置函数或组合用法。

本节课将系统性地讲解PySpark中最常用的内置函数，按照功能模块分为：字符串函数、日期时间函数、聚合函数、条件函数、窗口函数、集合函数等。每个函数都会配合实战代码和业务场景，帮助你彻底掌握并能在实际工作中灵活运用。学完这节课，你将不再需要为简单的数据转换编写UDF，代码性能将提升一个数量级。

二、学习目标

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

三、核心理论知识点

四、原理通俗讲解

4.1 内置函数为何比UDF快得多？

当你使用Python UDF时，例如：

@udf("string")
def my_upper(s):
return s.upper()

Spark在执行时需要：

这个过程涉及多次序列化和跨进程通信，开销极大。

而内置函数（如upper(col("name"))）完全在JVM内部执行，直接操作二进制数据（Tungsten格式），无需序列化，并且可以参与Catalyst优化器的表达式重写和代码生成，最终生成高度优化的Java字节码。

因此，能用内置函数解决的场景，绝对不要写UDF。

4.2 函数组合与链式调用

内置函数通常是表达式（Column类型），可以进行组合。例如：

df.withColumn("domain", split(regexp_extract(col("email"), "@(.*)", 1), "\\\\.").getItem(0))

这种链式调用在逻辑上构建了一棵表达式树，Catalyst优化器能够对这棵树进行常量折叠、谓词下推等优化。

4.3 窗口函数的物理执行

窗口函数需要将数据按照PARTITION BY分组，在每个分组内按照ORDER BY排序，然后计算排名或聚合。Spark会通过WindowExec物理算子来完成，它会对数据进行排序（可能触发Shuffle），然后在每个分区内逐行计算。合理使用窗口函数可以避免复杂的自连接。

五、重点概念拆解

5.1 字符串函数详解

注意：字符串索引从1开始（与SQL一致，与Python不同）。

5.2 日期时间函数详解

注意：to_date和to_timestamp在格式不匹配时会返回null，可以通过cast或try_cast（Spark 3.0+）处理。

5.3 聚合函数详解

5.4 条件函数详解

• when(condition, value).otherwise(default)：类似CASE WHEN，多个when可链式调用。
• coalesce(col1, col2, …)：返回第一个非空值。
• isnull(col)：判断是否为空，返回布尔值。
• if(cond, true_val, false_val)：类似三元运算符。
• nullif(expr1, expr2)：如果相等返回NULL，否则返回expr1。
• nanvl(col1, col2)：如果col1是NaN则返回col2，否则col1。

5.5 窗口函数详解

窗口函数语法：函数().over(Window.partitionBy("col1").orderBy("col2"))

常用窗口函数：

• row_number()：连续唯一编号，从1开始。
• rank()：有间隔排名（1,1,3,…）。
• dense_rank()：无间隔排名（1,1,2,…）。
• percent_rank()：相对排名 (rank-1)/(总行数-1)。
• ntile(n)：将分区内数据分成n个桶，返回桶编号。
• lag(col, offset, default)：向前偏移offset行。
• lead(col, offset, default)：向后偏移。
• 聚合函数如sum(col).over(…)：计算累计和、移动平均等。

窗口范围定义：

from pyspark.sql import Window
Window.partitionBy("user").orderBy("date").rowsBetween(Window.unboundedPreceding, Window.currentRow)

• rowsBetween(start, end)：按行数定义窗口范围。
• rangeBetween(start, end)：按排序值的差值定义。

5.6 集合函数（数组与Map）

处理ArrayType列的函数：

• array_contains(array, value)：是否包含。
• size(array)：数组长度。
• explode(array)：将数组每个元素展开成多行（一对多）。
• posexplode(array)：同时展开索引和元素。
• array_distinct(array)：去重。
• array_remove(array, value)：删除指定元素。
• array_union(array1, array2)：并集。
• array_intersect(array1, array2)：交集。
• array_join(array, delimiter)：将数组元素拼接为字符串。

处理MapType列的函数：

• map_keys(map)：返回所有键的数组。
• map_values(map)：返回所有值的数组。
• map_from_arrays(keys, values)：从两个数组创建Map。

六、易错点避坑

6.1 混淆substring的索引起始

SQL中substring(str, pos, len)的pos从1开始，而Python字符串切片从0开始。错误习惯会导致截取结果错误。

6.2 日期格式字符串大小写敏感

date_format中格式必须严格匹配：yyyy-MM-dd（小写y表示年，大写M表示月，小写m表示分钟）。常见错误：用YYYY（这是周年周数）导致年份错误。

6.3 窗口函数中order by缺少会导致整个分区作为一个窗口

如果不指定orderBy，窗口默认为整个分区（无滑动窗口）。但某些窗口函数（如row_number）要求必须指定orderBy，否则报错。

6.4 collect_list结果无序且可能受分区影响

collect_list不保证顺序，如果需要有序，应先orderBy再收集。

6.5 在聚合前使用explode导致数据膨胀

explode会为数组每个元素生成一行，如果数组很大，会大幅增加数据量。在groupBy前使用explode可能导致Shuffle数据量爆炸。

6.6 使用when时遗漏otherwise

如果没有otherwise，不满足条件的分支会返回null，这可能是期望的结果，但也可能无意中引入空值。

6.7 coalesce与repartition混淆

coalesce作为函数用于返回第一个非空值，而DataFrame的coalesce方法用于减少分区。注意上下文。

七、完整实战案例

本案例将使用一个模拟的用户行为日志数据集，综合运用字符串、日期、聚合、条件、窗口、集合等内置函数，完成以下分析任务：

# ============== builtin_functions_demo.py ==============
# 功能：PySpark内置函数全解实战（字符串、日期、聚合、条件、窗口、集合）
# 数据量：模拟100万条Apache日志
# 环境：PySpark 3.x

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql import Window
import random
from datetime import datetime, timedelta

# ========== 1. 创建SparkSession ==========
spark = SparkSession.builder \\
.appName("BuiltinFunctionsDemo") \\
.master("local[4]) \\
.config("spark.sql.shuffle.partitions", "8") \\
.getOrCreate()

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

print("=" * 80)
print("PySpark 内置函数全解实战")
print("=" * 80)

# ========== 2. 生成模拟日志数据 ==========
# 原始日志格式：IP – – [日期:时间 时区] "METHOD URL PROTOCOL" STATUS BYTES
# 示例：192.168.1.1 – – [01/Jan/2024:12:34:56 +0800] "GET /api/user HTTP/1.1" 200 1234
print("\\n生成模拟访问日志…")
ips = [f"192.168.1.{i}" for i in range(1, 101)]
urls = ["/api/user", "/api/order", "/api/product", "/home", "/login", "/static/style.css", "/api/order/create"]
methods = ["GET", "POST"]
status_codes = [200, 200, 200, 404, 500, 302]

def generate_log_line(timestamp):
ip = random.choice(ips)
method = random.choice(methods)
url = random.choice(urls)
status = random.choice(status_codes)
bytes_sent = random.randint(100, 10000)
log_line = f'{ip} – – [{timestamp.strftime("%d/%b/%Y:%H:%M:%S +0800")}] "{method} {url} HTTP/1.1" {status} {bytes_sent}'
return log_line

# 生成最近7天的日志，共100万条
num_records = 1_000_000
start_date = datetime.now() – timedelta(days=7)
logs = []
for i in range(num_records):
ts = start_date + timedelta(seconds=random.randint(0, 7*24*3600))
logs.append(generate_log_line(ts))

# 创建DataFrame（单列）
log_df = spark.createDataFrame([(log,) for log in logs], ["raw_log"])
print(f"生成了 {log_df.count():,} 条日志记录")
log_df.show(5, truncate=False)

# ========== 3. 字符串函数：解析日志 ==========
print("\\n" + "=" * 80)
print("步骤1: 使用字符串函数解析日志")
print("=" * 80)

# 定义正则表达式提取字段
parsed_df = log_df.select(
regexp_extract(col("raw_log"), r'^(\\S+)', 1).alias("ip"), # IP
regexp_extract(col("raw_log"), r'\\[(.*?)\\]', 1).alias("timestamp_str"), # 时间戳
regexp_extract(col("raw_log"), r'"(\\S+) (\\S+) \\S+"', 1).alias("method"), # 方法
regexp_extract(col("raw_log"), r'"\\S+ (\\S+) \\S+"', 1).alias("url"), # URL
regexp_extract(col("raw_log"), r'"\\S+ \\S+ \\S+" (\\d{3})', 1).cast("int").alias("status"), # 状态码
regexp_extract(col("raw_log"), r'"\\S+ \\S+ \\S+" \\d{3} (\\d+)', 1).cast("int").alias("bytes") # 字节数
)

# 使用字符串函数进一步清洗
# 去除URL中的查询参数（?后面的部分）
parsed_df = parsed_df.withColumn(
"url_path",
split(col("url"), "\\\\?").getItem(0)
)

# 对IP进行分类（IP段）
parsed_df = parsed_df.withColumn(
"ip_segment",
concat(split(col("ip"), "\\\\.").getItem(0), lit("."),
split(col("ip"), "\\\\.").getItem(1), lit("."),
split(col("ip"), "\\\\.").getItem(2))
)

print("解析后的DataFrame:")
parsed_df.printSchema()
parsed_df.show(10, truncate=False)

# ========== 4. 日期函数：转换时间戳、提取小时/星期几 ==========
print("\\n" + "=" * 80)
print("步骤2: 使用日期函数处理时间")
print("=" * 80)

# 将字符串时间戳转为timestamp类型，格式：01/Jan/2024:12:34:56 +0800
# 需要先去掉时区部分
parsed_with_date = parsed_df.withColumn(
"timestamp_clean",
regexp_replace(col("timestamp_str"), " \\\\+\\\\d{4}$", "")
).withColumn(
"log_time",
to_timestamp(col("timestamp_clean"), "dd/MMM/yyyy:HH:mm:ss")
).withColumn(
"date",
to_date(col("log_time"))
).withColumn(
"hour",
hour(col("log_time"))
).withColumn(
"day_of_week",
dayofweek(col("log_time")) # 1=Sunday, 7=Saturday
).withColumn(
"weekday",
when(dayofweek(col("log_time")).isin(1, 7), "Weekend").otherwise("Weekday")
)

print("日期处理后的Schema:")
parsed_with_date.printSchema()
parsed_with_date.select("timestamp_str", "log_time", "date", "hour", "weekday").show(5, truncate=False)

# ========== 5. 条件函数：状态码分类、空值处理 ==========
print("\\n" + "=" * 80)
print("步骤3: 使用条件函数 (when/otherwise, coalesce)")
print("=" * 80)

# 根据状态码分类
classified_df = parsed_with_date.withColumn(
"status_category",
when(col("status") < 300, "Success")
.when(col("status") < 400, "Redirect")
.when(col("status") < 500, "Client Error")
.otherwise("Server Error")
).withColumn(
"is_error",
when(col("status") >= 400, True).otherwise(False)
)

# 处理可能为空的bytes字段（使用coalesce）
classified_df = classified_df.withColumn(
"bytes_sent",
coalesce(col("bytes"), lit(0))
)

print("条件分类结果:")
classified_df.select("status", "status_category", "is_error", "bytes_sent").show(10)

# ========== 6. 聚合函数 + 分组 + 透视 ==========
print("\\n" + "=" * 80)
print("步骤4: 聚合函数与透视（pivot）")
print("=" * 80)

# 按小时统计PV、UV、平均响应时间
hourly_stats = classified_df.groupBy("date", "hour").agg(
count("*").alias("pv"),
approx_count_distinct("ip").alias("uv"),
avg("bytes").alias("avg_bytes")
).orderBy("date", "hour")

print("每小时访问统计:")
hourly_stats.show(10)

# 使用pivot：按状态类别统计每天的数量
pivot_df = classified_df.groupBy("date").pivot("status_category", ["Success", "Redirect", "Client Error", "Server Error"]).count()
print("每天各状态类别数量（透视表）:")
pivot_df.show(5)

# 使用collect_list收集每个小时的URL列表（用于后续分析）
url_list_per_hour = classified_df.groupBy("date", "hour").agg(
collect_list("url_path").alias("urls")
)
print("每小时URL列表（collect_list示例）:")
url_list_per_hour.show(3, truncate=False)

# ========== 7. 窗口函数：排名、累计、滞后 ==========
print("\\n" + "=" * 80)
print("步骤5: 窗口函数实战")
print("=" * 80)

# 窗口定义：按IP分区，按时间排序
window_spec = Window.partitionBy("ip").orderBy("log_time")

# 为每个IP的请求按时间编号
df_with_row_num = classified_df.withColumn(
"request_seq",
row_number().over(window_spec)
)

# 每个IP的累计响应字节数（累计和）
df_with_cumsum = classified_df.withColumn(
"cumulative_bytes",
sum("bytes").over(window_spec.rowsBetween(Window.unboundedPreceding, Window.currentRow))
)

# 每个IP上一次请求的时间（LAG）
df_with_lag = classified_df.withColumn(
"prev_log_time",
lag("log_time", 1).over(window_spec)
).withColumn(
"time_since_last",
datediff(col("log_time"), col("prev_log_time"))
)

# 每个IP的请求次数排名（全局窗口）
ip_rank_window = Window.orderBy(desc("cnt"))
ip_rank_df = classified_df.groupBy("ip").count().withColumn(
"rank",
rank().over(ip_rank_window)
).filter(col("rank") <= 10)

print("访问次数Top10的IP:")
ip_rank_df.show()

# 使用ntile将每个IP的请求分为5个桶（用于百分位数分析）
df_with_ntile = classified_df.withColumn(
"bytes_ntile",
ntile(5).over(Window.partitionBy("ip").orderBy("bytes"))
)

print("每个IP的请求按响应字节数分桶:")
df_with_ntile.select("ip", "bytes", "bytes_ntile").show(20)

# ========== 8. 集合函数（数组/Map） ==========
print("\\n" + "=" * 80)
print("步骤6: 集合函数（数组操作）")
print("=" * 80)

# 将URL路径按'/'拆分成数组，并分析路径深度
df_with_path = classified_df.withColumn(
"path_parts",
split(col("url_path"), "/")
).withColumn(
"path_depth",
size(col("path_parts")) – 1 # 减去空字符串
).withColumn(
"first_level",
when(size(col("path_parts")) > 1, col("path_parts").getItem(1)).otherwise(lit("root"))
)

# 使用explode展开路径组件，统计每个路径组件的访问频率
exploded_df = classified_df.withColumn(
"path_parts_array",
split(col("url_path"), "/")
).withColumn(
"part",
explode(col("path_parts_array"))
).filter(col("part") != "")

path_component_stats = exploded_df.groupBy("part").count().orderBy(desc("count"))
print("URL路径组件访问频率（Top10）:")
path_component_stats.show(10)

# 使用array_contains判断URL是否包含特定字符串
classified_df.withColumn(
"is_api_call",
array_contains(split(col("url_path"), "/"), "api")
).select("url_path", "is_api_call").show(5)

# 收集每个IP访问的所有URL（去重）并拼接
ip_unique_urls = classified_df.groupBy("ip").agg(
collect_set("url_path").alias("unique_urls")
).withColumn(
"urls_string",
array_join(col("unique_urls"), ", ")
)
print("每个IP访问的不同URL（前10个IP）:")
ip_unique_urls.show(10, truncate=False)

# ========== 9. 综合案例：用户行为漏斗分析 ==========
print("\\n" + "=" * 80)
print("步骤7: 综合案例 – 页面访问漏斗分析")
print("=" * 80)

# 假设我们关注的路径顺序：/home -> /api/product -> /api/order/create -> /api/order/pay
funnel_steps = ["/home", "/api/product", "/api/order/create", "/api/order/pay"]

# 为每个IP的访问路径按时间排序，并获取路径序列
funnel_df = classified_df.select("ip", "log_time", "url_path") \\
.withColumn("rn", row_number().over(Window.partitionBy("ip").orderBy("log_time")))

# 简化漏斗计算：统计每个IP是否按顺序访问了这些路径（使用窗口函数的lag）
# 更复杂可使用pivot或序列匹配，这里演示思路
print("漏斗分析（简化）: 统计每个步骤的独立访客数")
for i, step in enumerate(funnel_steps):
step_count = classified_df.filter(col("url_path") == step).select("ip").distinct().count()
print(f"步骤{i+1}: {step} – 访客数 {step_count}")

# ========== 10. 性能优化：比较UDF与内置函数 ==========
print("\\n" + "=" * 80)
print("步骤8: 性能对比 – 内置函数 vs UDF")
print("=" * 80)

from pyspark.sql.functions import udf
import time

# UDF版本：提取URL的第一级路径
def extract_first_level_udf(url):
if not url:
return ""
parts = url.split("/")
return parts[1] if len(parts) > 1 else ""

extract_first_level = udf(extract_first_level_udf, StringType())

# 内置函数版本
start_time = time.time()
df_builtin = classified_df.withColumn("first_level", split(col("url_path"), "/").getItem(1))
df_builtin.count()
builtin_time = time.time() – start_time

start_time = time.time()
df_udf = classified_df.withColumn("first_level", extract_first_level(col("url_path")))
df_udf.count()
udf_time = time.time() – start_time

print(f"内置函数耗时: {builtin_time:.2f} 秒")
print(f"UDF耗时: {udf_time:.2f} 秒")
print(f"内置函数比UDF快 {udf_time/builtin_time:.1f} 倍")

# ========== 11. 展示常用函数示例汇总 ==========
print("\\n" + "=" * 80)
print("附录: 常用函数快速参考")
print("=" * 80)
print("""
字符串: concat, concat_ws, length, upper, lower, trim, substring, split, regexp_extract, regexp_replace
日期: current_date, current_timestamp, to_date, to_timestamp, date_format, year, month, datediff, date_add
聚合: sum, avg, count, max, min, collect_list, collect_set, approx_count_distinct
条件: when, otherwise, coalesce, isnull, if
窗口: row_number, rank, dense_rank, lag, lead, ntile, sum().over()
集合: explode, posexplode, size, array_contains, array_distinct, array_join
""")

# ========== 12. 清理 ==========
spark.stop()
print("\\n✅ 内置函数演示完成")

八、代码逐行解析

8.1 正则提取字段

regexp_extract(col("raw_log"), r'^(\\S+)', 1).alias("ip")

使用正则捕获组提取IP，^(\\S+)匹配行首非空白字符序列。1表示返回第一个捕获组。类似地提取其他字段。注意状态码和字节数在提取后使用.cast("int")转换为整型。

8.2 字符串分割与取元素

split(col("url"), "\\\\?").getItem(0)

split返回数组，getItem(0)取第一个元素。注意正则表达式中的?需要转义。

8.3 日期转换

regexp_replace(col("timestamp_str"), " \\\\+\\\\d{4}$", "")

先去掉时区部分（如+0800），然后使用to_timestamp指定格式dd/MMM/yyyy:HH:mm:ss。注意月份是MMM（英文缩写）。hour、dayofweek等提取部分。

8.4 条件分类

when(col("status") < 300, "Success")
.when(col("status") < 400, "Redirect")
...

多个when链式调用，类似CASE WHEN。otherwise可选。

8.5 近似去重计数

approx_count_distinct("ip").alias("uv")

对于超大基数去重（如UV），使用approx_count_distinct可以提高速度，误差可接受（默认为5%）。

8.6 透视（pivot）

pivot("status_category", ["Success", "Redirect", ...])

将行转列，第二个参数指定列值列表（可选，但指定后性能更好）。

8.7 窗口函数

Window.partitionBy("ip").orderBy("log_time")
row_number().over(window_spec)

为每个IP的请求按时间编号。sum("bytes").over(window_spec.rowsBetween(Window.unboundedPreceding, Window.currentRow)) 计算累计和。

8.8 数组展开

explode(col("path_parts_array"))

将数组每个元素展开成一行，常用于日志解析中的标签拆分。

8.9 UDF vs 内置函数性能对比

内置函数直接操作JVM数据，而UDF需要序列化和进程间通信，耗时高出数十倍。因此强烈建议优先使用内置函数。

九、业务场景落地应用

9.1 场景一：用户注册日志中的设备信息解析

原始日志中用户代理字符串（User-Agent）需要提取操作系统、浏览器版本。使用regexp_extract和when组合。

df.withColumn("os",
when(col("ua").rlike("Windows"), "Windows")
.when(col("ua").rlike("Mac"), "macOS")
.when(col("ua").rlike("Linux"), "Linux")
.otherwise("Other")
).withColumn("browser",
regexp_extract(col("ua"), "(Chrome|Firefox|Safari|Edge)", 1)
)

9.2 场景二：电商订单的RFM分层

使用窗口函数对用户进行消费金额分桶，计算每个用户的排名。

rfm_df = order_df.groupBy("user_id").agg(
max("order_date").alias("last_order"),
count("*").alias("frequency"),
sum("amount").alias("monetary")
).withColumn("recency", datediff(current_date(), col("last_order")))

# 分桶打分
rfm_df.withColumn("r_score", ntile(5).over(Window.orderBy(col("recency").desc()))) \\
.withColumn("f_score", ntile(5).over(Window.orderBy(col("frequency")))) \\
.withColumn("m_score", ntile(5).over(Window.orderBy(col("monetary"))))

9.3 场景三：实时流处理中的JSON解析

使用from_json函数解析Kafka消息中的JSON字符串。

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([StructField("user_id", IntegerType()), StructField("action", StringType())])
df = df.withColumn("json", from_json(col("value"), schema)).select("json.*")

9.4 场景四：日志路径分析中的路径拆分

分析用户访问路径，统计每个页面的深度。

df.withColumn("depth", size(split(col("url"), "/")) – 1) \\
.groupBy("depth").count().orderBy("depth")

十、常见报错排查

10.1 to_timestamp返回null

原因：日期字符串格式不匹配，或时区问题。

解决：使用date_format先转换为标准格式，或尝试多种格式（Spark 3.0+支持to_timestamp自动解析常见格式）。

10.2 窗口函数报错Window function without frame

原因：某些函数（如sum over窗口）需要指定范围（rowsBetween）或默认范围，但若同时使用order by而没有指定范围，Spark会使用默认范围（range between unbounded preceding and current row），某些情况下可能出错。对于排名函数不需要范围。

解决：对于聚合窗口函数，显式指定rowsBetween。

10.3 regexp_extract提取不到值返回空字符串

原因：正则未匹配，返回空字符串而非null。

解决：使用when判断长度或结合regexp检查。

10.4 使用collect_list导致内存溢出

原因：分组内数据量太大，数组占用内存过大。

解决：限制每组大小，或使用approx_collect_list（第三方库），或改用其他聚合方式。

10.5 explode后数据膨胀导致Shuffle数据量巨大

原因：每行数组长度很大，explode后行数乘倍数增加。

解决：在explode前先过滤掉空数组，或考虑其他设计避免爆炸。

十一、本节课知识点总结

函数分类速查表

性能优化要点

十二、课后思考作业

作业一：理论理解题

作业二：代码实践题

作业三：场景应用题

某资讯App的用户点击日志包含字段：user_id, article_id, event_time, read_duration（阅读时长秒）。需要分析：

• 用户阅读时长的分布（按分钟分桶）
• 每个用户最后一次阅读的文章ID（使用窗口函数）
• 每天阅读时长最长的前10篇文章
• 将文章ID列表按用户聚合，并输出每个用户阅读过的不同文章数（去重）

请使用内置函数实现上述分析，并说明关键步骤使用的函数。

作业四：拓展研究

提交方式：本次作业要求提交代码脚本和运行结果截图，以及理论题的答案。鼓励将常用的内置函数封装成工具类。

扩展阅读：

• PySpark官方函数文档：pyspark.sql.functions
• 《Spark SQL 性能优化》第6章
• 开源项目：pyspark-builtin-functions-cheatsheet

通过本节课的学习，你已经掌握了PySpark内置函数的精髓，能够高效地进行数据转换和分析。在实际工作中，遇到数据处理需求，应第一时间想到内置函数，而不是UDF。下一节课我们将学习PySpark自定义函数（UDF、UDAF、UDTF）的开发原理与企业实战，了解在必须使用自定义逻辑时如何写出高性能的扩展函数。我们下节课见！

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

去订阅

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