当前位置： 首页 > news >正文

PySpark groupBy 原理与高可用实践：从数据倾斜到AQE调优

news 2026/5/26 5:58:57

1. 为什么 PySpark 的 groupBy 是大数据聚合的“心脏”，而不是一个普通函数？

在真实的大数据生产环境里，我见过太多团队把 PySpark 当成“加强版 Pandas”来用——写完groupBy就立刻.collect()，结果集群内存爆满、任务卡死在 Stage 3、运维半夜被电话叫醒。这不是 Spark 的问题，而是没理解groupBy在整个计算引擎里的真实角色：它不是一句 SQL 的简单翻译，而是触发分布式计算流水线启动的“点火开关”。

你手里的那行df.groupBy("department").sum("salary")，表面看只是分组求和，背后却是一整套精密协作的工程系统在运转。它要决定数据怎么切、键怎么分、中间结果存哪、失败了怎么重试、倾斜了怎么兜底。这就像你按下车钥匙上的“启动”按钮，听到引擎轰鸣，但真正让车跑起来的，是燃油喷射、点火正时、变速箱换挡逻辑这一整套底层机制。

所以，这篇文章不叫“PySpark groupBy 用法详解”，而叫“Mastering PySpark’s groupBy”。Mastering 的意思是：你得知道它什么时候该用、为什么这么设计、哪里容易卡壳、出了问题怎么一眼定位。它解决的从来不是“怎么写出来”，而是“怎么让它在 TB 级数据、百节点集群上，稳定、快速、可预测地跑完”。

核心关键词就三个：可扩展性（Scalable）、确定性（Deterministic）、可观测性（Observable）。

  • 可扩展性，指它能从本地单机测试的 10 行数据，无缝扩展到集群上处理每天 50TB 的日志；
  • 确定性，指同样的代码、同样的数据，在开发环境、测试环境、生产环境跑出来的结果必须完全一致，不能今天对、明天错；
  • 可观测性，指当它变慢或出错时，你能通过 Spark UI、执行计划、日志三分钟内锁定瓶颈，而不是靠猜和重启。

这三点，恰恰是传统单机分析工具（比如 Pandas、Excel）天然不具备，而企业级数据平台又绝对依赖的。我带过的两个金融风控项目，上线后第一周就因groupBy倾斜导致 T+1 报表延迟 6 小时，最后发现是“未知客户”这个分类占了全量数据的 73%，所有计算都挤在一个 executor 上。这种问题，光会写语法根本没用，必须懂它的运行肌理。

所以，别把它当成一个 API 来学，要当成一个“分布式计算契约”来理解：你告诉 Spark 你想按什么分组、算什么指标，Spark 就承诺给你一个符合 ACID 基础语义（至少是 at-least-once）、能水平扩展、失败可重试的结果。而你的责任，是提供清晰、无歧义、可分区的分组键，以及避免引入破坏这个契约的操作（比如在 groupBy 后用 Python for 循环遍历结果）。

这也是为什么本文开篇就强调“Mastering”——因为一旦你只停留在“能跑通”的层面，当数据量翻 10 倍、业务逻辑加一层嵌套、集群资源临时紧张时，那个曾经很稳的 job，就会变成你监控告警列表里最常亮的那个红点。

2. groupBy 的底层设计哲学：Split-Apply-Combine 不是口号，是救命指南

很多初学者看到groupBy就想到 SQL 的GROUP BY，觉得“哦，就是分组嘛”，然后直接套用。结果一跑生产就懵：明明本地 10 万行秒出结果，线上 1 亿行跑了 40 分钟还 OOM。问题不在数据量，而在没吃透groupBy背后的Split-Apply-Combine（拆分-应用-合并）范式。这不是一个教学概念，而是 Spark 调度器做物理执行计划的唯一依据，更是你写高效代码的黄金法则。

2.1 Split（拆分）：分组键的选择，决定了 80% 的性能上限

Split 阶段，Spark 要把原始数据打散，把相同分组键（key）的数据强行“拉”到同一个 executor 上。这个过程叫Shuffle，它是整个groupBy流程里最昂贵、最不可控、也最容易出问题的环节。

关键点在于：分组键的分布质量，直接决定 shuffle 的均匀程度。

  • 如果你用"user_id"分组，而这个字段是 UUID，那么数据天然均匀，每个 partition 大概分到 1/200（默认 shuffle partitions 数）的数据，各 worker 干活很平均；
  • 但如果你用"status"分组，而这个字段只有["active", "inactive", "pending"]三个值，且active占了 95%，那 95% 的数据会被 shuffle 到同一个 partition，其他 199 个 partition 几乎空转——这就是典型的数据倾斜（Data Skew）。

我去年帮一家电商公司优化订单分析 pipeline，他们用"province"分组统计 GMV，结果发现广东、浙江、江苏三省占了全国订单的 62%。groupBy一跑，一个 task 跑了 22 分钟，其他 199 个 task 加起来才 3 分钟。根本不是代码问题，是分组维度本身就不适合做全局聚合。

实操心得：

提示：在写groupBy前，务必先用df.select("your_group_col").distinct().count()和df.groupBy("your_group_col").count().orderBy("count", ascending=False).show(10)快速探查分组键的基数（cardinality）和分布。如果 top1 的 count 占总数 > 30%，就要警惕倾斜风险，考虑加盐（salting）或预过滤。

2.2 Apply（应用）：聚合函数不是“算数”，而是“定义计算契约”

Apply 阶段，Spark 对每个分组内的数据执行你指定的聚合逻辑。这里最大的误区，是认为sum()、avg()这些函数只是“把数字加起来”。实际上，它们是 Spark 向你承诺的计算契约：

  • sum()承诺：对数值列做可交换、可结合的加法运算，支持 partial aggregation（先局部加，再全局加）；
  • count()承诺：对任意类型列做计数，且 null 值不参与计数；
  • collect_list()承诺：把分组内所有值收集到一个数组里，但它不支持 partial aggregation，必须把所有数据拉到一个节点才能执行——这就是为什么它极易 OOM。

这就引出了一个硬性原则：优先使用支持 partial aggregation 的内置函数。

  • ✅ 推荐：sum,count,avg,min,max,stddev,approx_count_distinct
  • ⚠️ 慎用：collect_list,collect_set,first,last（除非你 100% 确认分组内数据量极小）
  • ❌ 避免：自定义 UDF 做聚合（如@udf def my_sum(x): return sum(x)），因为它彻底绕过 Catalyst 优化，强制全量 shuffle + 单点计算。

我见过最惨的案例，是某推荐系统团队用collect_list("item_id")统计用户历史行为，结果一个高活用户有 20 万条记录，直接把 executor 内存撑爆。后来改成approx_count_distinct("item_id")+count("item_id")，性能提升 17 倍，结果误差在业务可接受范围内。

2.3 Combine（合并）：结果形态决定下游所有操作的成本

Combine 阶段，Spark 把各个 executor 计算出的中间结果（partial result）汇总，生成最终的 DataFrame。这个阶段输出的结构，会像影子一样影响你后续每一步操作。

例如：

  • df.groupBy("dept").sum("salary")输出两列：dept,sum(salary)；
  • df.groupBy("dept").agg(sum("salary").alias("total"), count("*").alias("cnt"))输出三列：dept,total,cnt；
  • df.groupBy("dept").pivot("role").sum("salary")输出动态列：dept,engineer,manager,intern……

关键洞察是：Combine 的输出是一个全新的、独立的 DataFrame，与原始 df 完全无关。它有自己的 schema、自己的分区策略、自己的执行计划。这意味着：

  • 如果你紧接着要filter("total > 10000")，Spark 会在 combine 后的新 DataFrame 上执行 filter，这是高效的；
  • 但如果你错误地写成df.filter("salary > 10000").groupBy("dept").sum("salary")，那就是先 filter 再 groupBy，虽然结果一样，但数据量可能差一个数量级——前者 shuffle 1 亿行，后者可能只 shuffle 200 万行。

避坑技巧：

注意：永远把filter放在groupBy前面，除非业务逻辑强制要求“先聚合再筛选”。比如“找出员工数 > 100 的部门”，必须先groupBy().count()再filter()；但“统计薪资 > 10K 的员工在各部门的总薪资”，就必须先filter()再groupBy()。顺序错了，成本天壤之别。

3. 核心实操：从零搭建一个抗压、可调、易排障的 groupBy 流程

纸上谈兵不如一次完整实操。下面我带你走一遍我在生产环境部署一个日活用户留存分析 job 的全过程。这个 job 每天处理 8.2 亿条 App 埋点日志，groupBy是其核心，我们不仅要让它跑得快，更要让它跑得稳、看得清、改得动。

3.1 环境准备与基础验证：别跳过这 3 分钟，它能省你 3 小时

首先，确认你的 SparkSession 配置不是默认的“玩具模式”。生产环境必须显式设置关键参数：

from pyspark.sql import SparkSession from pyspark.sql import functions as F spark = SparkSession.builder \ .appName("daily_retention_analysis") \ .config("spark.sql.adaptive.enabled", "true") \ # 开启自适应查询执行（AQE） .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \ # AQE 自动合并小分区 .config("spark.sql.adaptive.skewJoin.enabled", "true") \ # AQE 自动处理 join 倾斜 .config("spark.sql.shuffle.partitions", "128") \ # 根据集群规模调整，默认200太大 .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \ # 更快的序列化 .getOrCreate() # 强制验证：检查是否真的生效 print("Shuffle partitions:", spark.conf.get("spark.sql.shuffle.partitions")) print("AQE enabled:", spark.conf.get("spark.sql.adaptive.enabled"))

为什么这步不能跳？

  • spark.sql.shuffle.partitions=128：我们集群有 32 个 executor，每个配 4 core，128 是 32*4 的整数倍，能保证负载均衡。设成 200 会导致部分 executor 闲着，部分超载；
  • AQE=true：这是 Spark 3.0+ 的革命性特性，它能在运行时动态优化 shuffle 分区数、自动处理数据倾斜、合并小文件。不开它，等于放弃一半性能红利；
  • Kryo 序列化：比默认 Java 序列化快 10 倍，尤其对复杂对象（如嵌套 struct）效果显著。

3.2 数据探查与倾斜预判：用 5 行代码看清数据本质

假设我们有一张events表，包含user_id,event_date,event_type,app_version等字段。目标是计算“次日留存率”：当日新增用户中，第二天还回来的比例。

第一步，绝不是写 groupBy，而是用以下 5 行代码做“CT扫描”：

# 1. 查看数据总量和分区数 print("Total records:", events.count()) print("Num partitions:", events.rdd.getNumPartitions()) # 2. 探查 user_id 分布（核心分组键！） user_dist = events.select("user_id").groupBy("user_id").count() \ .orderBy("count", ascending=False).limit(5) user_dist.show() # 输出示例： # +--------------------+-----+ # | user_id|count| # +--------------------+-----+ # |00000000-0000-00...| 1245| # |11111111-1111-11...| 987| # |22222222-2222-22...| 876| # |33333333-3333-33...| 765| # |44444444-4444-44...| 654| # +--------------------+-----+ # 3. 计算倾斜率：top1 count / total count top1_ratio = user_dist.first()["count"] / events.count() print(f"Skew ratio: {top1_ratio:.4f}") # 如果 > 0.05，需警惕

实操心得：

提示：如果top1_ratio > 0.05（即前 1 名用户占了 5% 以上事件），说明user_id有轻度倾斜。这时不要急着加盐，先看业务：这个用户是不是测试账号、爬虫 ID 或异常设备？如果是，直接filter("user_id not in ('test_123', 'crawler_456')")预清洗，比技术手段更干净。

3.3 构建健壮的 groupBy 流程：从新增识别到留存计算的完整链路

现在开始写核心逻辑。注意，我们不是写一个函数，而是构建一个可审计、可复现、可中断恢复的 pipeline：

from pyspark.sql.window import Window from pyspark.sql.functions import col, lit, when, row_number, sum as spark_sum, count as spark_count # Step 1: 识别每日新增用户（按首次出现 event_date） # 使用 window function 找每个 user_id 的最小 event_date window_spec = Window.partitionBy("user_id").orderBy("event_date") first_event = events.withColumn("first_date", F.min("event_date").over(window_spec)) \ .filter(col("event_date") == col("first_date")) \ .select("user_id", "event_date").withColumnRenamed("event_date", "install_date") # Step 2: 关联次日行为（left join + date add） from pyspark.sql.functions import date_add retention_base = first_event.alias("i") \ .join( events.alias("e"), (col("i.user_id") == col("e.user_id")) & (col("e.event_date") == date_add(col("i.install_date"), 1)), "left" ) \ .select( col("i.user_id"), col("i.install_date"), col("e.event_date").isNotNull().alias("returned_next_day") ) # Step 3: 核心 groupBy —— 按 install_date 分组，统计新增数和次日返回数 # 关键：使用 agg() 一次性完成，避免多次 groupBy retention_result = retention_base \ .groupBy("install_date") \ .agg( spark_count("user_id").alias("new_users"), # 当日新增 spark_sum(when(col("returned_next_day"), 1).otherwise(0)).alias("returned") # 次日返回数 ) \ .withColumn("retention_rate", F.round(col("returned") / col("new_users") * 100, 2)) \ .orderBy("install_date") # Step 4: 缓存中间结果（因为后续可能多处引用） retention_result.cache() retention_result.count() # 触发 cache

为什么这样写？

  • window function找首次事件，比groupBy().min()更准，且能保留原始行；
  • left join+date_add实现“次日关联”，逻辑清晰，可读性强；
  • agg()里用when().otherwise()做条件计数，比先filter().count()再union更高效；
  • cache()放在最后，因为retention_result是最终指标，会被报表、告警、下游 ETL 多次消费。

3.4 性能调优实战：从 22 分钟到 3 分钟的 7 倍提速

上面的代码在测试环境跑得不错，但上线首日，retention_result.count()耗时 22 分钟。用explain(True)看物理计划，发现关键瓶颈：

== Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- HashAggregate(keys=[install_date#123], functions=[count(user_id#456), sum(cast(when...))]) +- Exchange hashpartitioning(install_date#123, 128) <-- 这里 shuffle 了 8.2 亿行！ +- HashAggregate(keys=[install_date#123], functions=[partial_count(user_id#456), partial_sum(...)]) +- Project [install_date#123, user_id#456, returned_next_day#789] +- BroadcastHashJoin ... <-- 这里用了 broadcast，但右表太大，broadcast 失败回退到 shuffle

优化动作：

  1. 预 repartition：在 join 前，把first_event按user_id重新分区，让 join 更高效：

    first_event_repart = first_event.repartition(200, "user_id") # 200 是 shuffle partitions 数

  2. 强制 AQE 处理倾斜：给retention_base加 hint，让 AQE 知道user_id可能倾斜：

    from pyspark.sql.functions import hints retention_base_hinted = hints.coalesce(200, retention_base) # 建议 AQE 合并分区

  3. 调整 shuffle 分区：针对这个 job，把spark.sql.shuffle.partitions临时设为 256（因为数据量大，128 不够）：

    spark.conf.set("spark.sql.shuffle.partitions", "256")

  4. 启用 AQE skew join（已配置，无需代码）。

优化后，物理计划里Exchange消失了，变成了BroadcastHashJoin，总耗时降到 3 分钟 12 秒。提速 7 倍的核心，不是算法，而是让数据在正确的时间、以正确的形态，出现在正确的节点上。

4. 高级聚合模式：rollup、cube、groupingSets 不是炫技，是业务需求的自然延伸

当你的分析从“部门总薪资”升级到“集团-大区-城市-门店”四级穿透，或者从“销售总额”扩展到“按产品线、按渠道、按时间”的任意组合分析时，groupBy就显得力不从心了。这时候，rollup、cube、groupingSets不是高级功能，而是业务语言的直接映射。

4.1 rollup：层级钻取的“金字塔”模型

rollup的本质，是按你指定的列顺序，从左到右逐级向上汇总，生成一个天然的层级结构。它完美匹配“集团→大区→城市→门店”这类树状管理架构。

以我们的样例数据为例，假设我们有region,city,store_id,revenue四列：

# 原始数据（简化） data = [ ("North", "Beijing", "N-BJ-001", 10000), ("North", "Beijing", "N-BJ-002", 12000), ("North", "Shanghai", "N-SH-001", 8000), ("South", "Guangzhou", "S-GZ-001", 15000), ] df = spark.createDataFrame(data, ["region", "city", "store_id", "revenue"]) # rollup 按 region -> city -> store_id 顺序 result = df.rollup("region", "city", "store_id") \ .agg(F.sum("revenue").alias("total_revenue")) \ .orderBy("region", "city", "store_id") result.show()

输出解读（关键！）：

+------+--------+--------+-------------+ |region| city|store_id|total_revenue| +------+--------+--------+-------------+ | NULL| NULL| NULL| 45000| <-- 全集团总计 | North| NULL| NULL| 30000| <-- 北区总计 | North|Beijing| NULL| 22000| <-- 北京市总计 | North|Beijing|N-BJ-001| 10000| <-- 具体门店 | North|Beijing|N-BJ-002| 12000| | North|Shanghai| NULL| 8000| | North|Shanghai|N-SH-001| 8000| | South| NULL| NULL| 15000| <-- 南区总计 | South|Guangzhou| NULL| 15000| | South|Guangzhou|S-GZ-001| 15000| +------+--------+--------+-------------+

业务价值：

  • BI 工具拖拽时，用户点“北区”，自动下钻到“北京市”，再点“北京市”，下钻到两家门店——rollup的 NULL 值就是 BI 工具识别层级的标记；
  • 财务月报，需要同时输出“全集团”、“各区域”、“各城市”三级数据，一份 SQL 就搞定，不用写三个groupBy。

4.2 cube：任意组合的“全息图谱”

如果说rollup是一条主干道，cube就是这张主干道上所有可能的交叉路口。它会生成你指定列的所有排列组合（2^n 种），包括单列、双列、三列……直到全列的聚合。

继续用上面的数据：

# cube 同样三列 result_cube = df.cube("region", "city", "store_id") \ .agg(F.sum("revenue").alias("total_revenue")) \ .orderBy("region", "city", "store_id") result_cube.show()

输出会多出这些行：

| NULL|Beijing| NULL| 22000| <-- 所有北京的店（不管哪个区） | NULL|Shanghai| NULL| 8000| | NULL|Guangzhou| NULL| 15000| | NULL| NULL|N-BJ-001| 10000| <-- 所有叫 N-BJ-001 的店（不管哪个区/市） | NULL| NULL|N-BJ-002| 12000| ...

业务场景：

  • 市场部想分析：“iPhone 在华东的销量”、“iPhone 在 2023 年的销量”、“华东在 2023 年的销量”，这三个维度任意两两组合的需求，cube一次产出，BI 直接切片；
  • 数据科学家做特征工程，需要所有可能的(product, region),(product, year),(region, year)组合统计，cube是最简洁的方案。

4.3 groupingSets：精准控制的“定制化聚合”

groupingSets是最灵活的，它让你像写 SQL 的GROUPING SETS一样，精确指定你要哪几组聚合，不多不少。

回到最初的人事数据，如果我们只要求：

  • 按department和employee的明细（A）
  • 按department的汇总（B）
  • 全公司的总计（C）

那么：

from pyspark.sql.functions import grouping_id result_gs = df.groupingSets( [ # 明确列出三组 ("department", "employee"), # A: 部门+员工 ("department", ), # B: 仅部门 () # C: 全公司（空元组） ], "department", "employee" # 这里声明所有可能用到的列，供 grouping_id 使用 ).agg( F.sum("salary").alias("total_salary") ).withColumn("grouping_level", F.grouping_id()) \ .orderBy("department", "employee") result_gs.show()

输出中的grouping_level是关键：

  • grouping_level=0：表示("department", "employee")这组（即明细行）；
  • grouping_level=1：表示("department", )这组（即部门汇总，employee为 NULL）；
  • grouping_level=3：表示()这组（即全公司，department和employee都为 NULL）。

为什么用 groupingSets 而不是 cube？

  • cube会生成 2^2=4 组（dep+emp,dep,emp,all），但我们不需要单独的employee汇总；
  • groupingSets只生成你要的 3 组，数据量更小，查询更快，语义更清晰。在千万级数据上，cube可能比groupingSets慢 30%。

5. 故障排查与避坑指南：那些让你凌晨三点还在看 Spark UI 的真实问题

再完美的设计，也会遇到生产环境的“惊喜”。以下是我在过去三年里，从上百个groupBy相关故障中总结出的Top 5 致命问题清单，附带一键诊断命令和根治方案。

5.1 问题 1：Stage 卡死在 “Shuffle Write” 阶段，executor 内存持续上涨

现象：
Spark UI 中，某个 Stage 的 Task Duration 显示 “Running”，但 Shuffle Write Bytes 一直涨，GC Time 占比 > 80%，executor 日志里反复出现java.lang.OutOfMemoryError: Java heap space。

根因：
这是最经典的数据倾斜。某个分组键（如user_id="crawler_123"）的数据量远超其他，导致一个 task 要处理 GB 级数据，而 JVM 堆内存不够。

一键诊断：
在 Spark UI 的 “Stages” 页面，点击卡住的 Stage，看 “Task Summary” 下的 “Shuffle Write Size” 列。如果某一行的值是其他行的 100 倍以上，就是它。

根治方案：
加盐（Salting）—— 最通用、最有效的方案。

from pyspark.sql.functions import rand, concat, lit, col # 步骤1：给分组键加一个随机后缀（盐） salted_df = df.withColumn("salted_key", concat(col("user_id"), lit("_"), (rand() * 10).cast("int"))) # 步骤2：按 salted_key 分组（此时数据均匀了） salted_grouped = salted_df.groupBy("salted_key").agg(F.sum("revenue")) # 步骤3：去掉盐，按原始 key 汇总（此时每个原始 key 对应多个 salted_key，但数据量小了） final_result = salted_grouped \ .withColumn("original_key", F.split(col("salted_key"), "_")[0]) \ .groupBy("original_key") \ .agg(F.sum("sum(revenue)").alias("total_revenue"))

为什么有效？

  • rand() * 10把一个热 key 拆成 10 个新 key，分散到 10 个 task；
  • 最后groupBy("original_key")的数据量是1/10，不再倾斜；
  • 盐值范围（10）要根据倾斜程度调，一般 10-100。

5.2 问题 2：groupBy结果为空，但原始数据明明有数据

现象：
df.show()能看到数据，df.count()返回 1000，但df.groupBy("col").count().show()一行不显示，count()返回 0。

根因：
col列里全是NULL值。groupBy默认会把NULL当作一个分组键，但如果所有值都是NULL，count()会返回 0（因为count(*)不统计 NULL，而count("col")也不统计 NULL）。

一键诊断：

# 查看该列的 NULL 率 null_ratio = df.select(F.mean(F.col("col").isNull().cast("int"))).first()[0] print(f"NULL ratio: {null_ratio}") # 如果为 1.0，就是全 NULL

根治方案：

  • 业务层：修复上游数据源，确保col有有效值；
  • 代码层：在groupBy前na.drop(subset=["col"])或na.fill(value="UNKNOWN", subset=["col"])。

5.3 问题 3：agg()里用collect_list导致 OOM，但业务强需求

现象：
需要输出每个部门的员工姓名列表，但collect_list("name")一跑就内存溢出。

根因：
collect_list不支持 partial aggregation，必须把一个分组内所有数据拉到一个 executor 内存里。

根治方案：
用approx_count_distinct+sample替代，或改用collect_set。

# 方案1：如果只需要去重后的名字（不关心重复次数） df.groupBy("department").agg(F.collect_set("name").alias("names")).show() # 方案2：如果必须全量，且数据量可控，用 sample 采样 df.sample(0.1).groupBy("department").agg(F.collect_list("name").alias("sampled_names")).show() # 方案3（推荐）：用窗口函数 + limit，只取前 N 个 from pyspark.sql.window import Window window_spec = Window.partitionBy("department").orderBy(F.col("salary").desc()) df.withColumn("rn", F.row_number().over(window_spec)) \ .filter("rn <= 10") \ .groupBy("department") \ .agg(F.collect_list("name").alias("top10_names")) \ .show()

5.4 问题 4：groupBy后filter不生效，或结果不符合预期

现象：
df.groupBy("dept").agg(F.sum("salary")).filter("sum(salary) > 10000").show()返回空，但手动算过有部门超 10000。

根因：
filter()里写的列名sum(salary)是 Spark 自动生成的，但实际列名可能是sum(salary)#123L。直接写字符串会解析失败。

根治方案：
永远用列对象（column object）做 filter。

# ✅ 正确：用 agg 返回的列对象 agg_df = df.groupBy("dept").agg(F.sum("salary").alias("total_salary")) agg_df.filter(agg_df.total_salary > 10000).show() # ✅ 或者用 col() from pyspark.sql.functions import col agg_df.filter(col("total_salary") > 10000).show() # ❌ 错误：用字符串 agg_df.filter("total_salary > 10000").show() # 可能失效

5.5 问题 5：groupBy在本地跑得飞快，集群上慢如蜗牛

现象：
本地pyspark --master local[4]10 秒跑完，YARN 集群上 10 分钟。

根因：
本地模式没有网络 shuffle，所有计算都在内存里完成；集群模式必须走网络，而你的集群网络或磁盘 I/O 是瓶颈。

一键诊断：
在 Spark UI 的 “Storage” 页面，看 “Disk Spill” 是否 > 0。如果Shuffle spill (memory)和Shuffle spill (disk)都很大，说明 executor 内存不足，频繁 GC 和磁盘 IO。

根治方案：
调大 executor 内存，并开启 shuffle 压缩。

# 提交 job 时 spark-submit \ --executor-memory 8g \ --conf spark.shuffle.compress=true \ --conf spark.shuffle.spill.compress=true \ your_job.py

6. DataFrame vs RDD：为什么你几乎永远不该用 RDD 的 groupByKey

这个问题我被问过不下 50 次：“RDD 的groupByKey和 DataFrame 的groupBy，到底哪个快？” 答案很明确：在 99.9% 的场景下，DataFramegroupBy快得多，且更稳定、更易维护。用 RDDgroupByKey，不是“选择”，而是“降级”。

6.1 性能对比：不只是快，是质的飞跃

让我们用同一份数据（1000 万行，user_id,amount）做实测：

| 操作 | DataFrame groupBy | RDD groupByKey | RDD reduceByKey | |------|------------------|

http://www.jsqmd.com/news/888313/

相关文章：

  • 基于TypeScript与NeuroLink构建企业级AI代理：架构设计与实战指南
  • Android应用安全防护核心技术深度剖析：加壳技术详解与实战
  • Unity里别再只会用Parent了！试试Constraint组件，动态绑定物体更灵活
  • 告别SD卡！手把手教你为EBAZ4205矿卡配置NAND启动的JFFS2根文件系统（Petalinux 2018.3）
  • 别再只盯着大模型了，2026年真正拉开AI体验差距的是资料后勤系统
  • VR与机器学习如何为神经多样性群体构建个性化安全训练沙盒
  • 手把手教你用迅雷搞定USRP固件下载，让GNUradio在Linux上跑起来
  • 告别飞线乱麻！用立创EDA的布局传递与模块化思维高效规划你的PCB
  • 目视初检+万用表快测，PCB元件损坏快速定位法
  • 【面试必备】面试官问你“理解架构吗？“的标准答案
  • 告别外设不足：用MCP2517FD给ESP32或树莓派Pico扩展CAN FD接口实战
  • 2026年热门的衡水可多次注浆管/衡水桩基注浆管厂家哪家好 - 行业平台推荐
  • 从‘纹波’看本质：手把手教你诊断并优化VNA去嵌后的S参数测量结果
  • Unity PC单exe封装实战：嵌入式资源方案详解
  • Unity打包安卓报错？手把手教你修改build.gradle解决资源冲突（附Gradle模板配置）
  • 避坑指南：MPU6050 DMP采样率配置的那些“坑”与最佳实践
  • 21.开源万能刷机工具！跨 Windows/Linux/macOS，支持安卓 + 苹果全机型
  • 交通流预测模型对比：从短期精准到长期稳健的选型指南
  • 别再死记硬背公式了！用Multisim 14.0仿真文件，带你玩转20个经典运放电路
  • Excel饼图说服力设计：从视觉认知到业务决策
  • C#游戏物理引擎的SIMD向量加速实战
  • 精通 Android NDK/JNI：从入门到精通实战与面试精粹
  • Promptfoo实战：构建可版本化、自动化的LLM输出质量评估体系
  • 4-20mA回路供电显示模块设计：低功耗高精度工业仪表方案
  • 终极指南：如何用开源分屏工具实现单机游戏多人同乐
  • 手把手教你：如何根据你的CH32芯片型号（F103/V103）正确设置WCH-Link下载模式
  • ComfyUI-WanVideoWrapper架构设计与企业级视频生成实现原理
  • 别再写重复代码了！用这个Spine动画管理器搞定Unity中的角色动作切换与回调
  • 配置 OpenClaw 使用 Taotoken 作为其大模型供应商
  • 低碳物流网络设计与评价【附代码】