窗口函数并行度由spark.sql.shuffle.partitions决定，与window()内参数无关；应设为集群总core数（如120），高基均匀键可设2–3倍，倾斜场景需先盐化再调优。
窗口函数本身不直接控制并行度，真正起作用的是底层 shuffle 分区数 —— 也就是 spark.sql.shuffle.partitions 的值。 直接在 window() 或 over() 里加参数、改 orderBy 字段顺序、甚至用 rangeBetween 都不会改变 task 数量。窗口计算一旦触发 shuffle（比如按 partitionBy 分组 + 排序），就会走这个全局 shuffle 分区配置。
为什么窗口操作容易卡住或慢得离谱
常见现象是：SQL 里用了 row_number() over (partition by user_id order by ts)，数据量中等（几亿行），但 stage 卡在 shuffle write/read，executor CPU 利用率低，GC 频繁，日志里反复出现 ShuffleMapStage 等待。
- 根本原因不是窗口逻辑写错了，而是默认
spark.sql.shuffle.partitions=200和你集群实际 core 数严重不匹配 - 比如 YARN 上跑了 30 个 executor，每个 4 core → 总共 120 个可用 core，但 shuffle 分区设成 200，意味着平均每个 core 要处理近 2 个 partition，中间文件多、网络拉取压力大、磁盘 spill 高
- 更糟的是，如果分区数远小于 core 数（比如只设 50），大量 core 空转，资源浪费且单 task 处理数据量暴增，OOM 或 GC 拖垮整个 stage
怎么设才对：优先级与生效时机
窗口操作的 shuffle 并行度只认 spark.sql.shuffle.partitions，它和其他并行度参数互不影响，也不受 RDD 的 spark.default.parallelism 控制。关键点是：代码里 .config("spark.sql.shuffle.partitions", "N") 的设置会覆盖提交时 --conf 和 spark-defaults.conf 里的同名配置。
- 推荐在创建
SparkSession时就固定设好，例如：val spark = SparkSession.builder() .config("spark.sql.shuffle.partitions", "120") // = executor 数 × core 数 .getOrCreate() - 如果用 SQL 方式执行（
spark.sql("select row_number() over (...)")），这个配置必须提前生效；运行时用spark.conf.set("spark.sql.shuffle.partitions", "120")也有效，但仅对后续 SQL 生效 - 不要试图用
repartition(120)在窗口前强行打散数据 —— 这会引入额外 shuffle，反而增加开销；窗口算子自己就会 shuffle，重复 shuffle 得不偿失
哪些情况要调小，哪些必须调大
不是越大越好，也不是越接近 core 数就越优。窗口操作的 shuffle 压力集中在 partitionBy 键的分布上，所以还要看 key 的基数和倾斜程度。
- 当
partitionBy字段高基（如user_id有上亿不同值）且均匀：分区数可设为总 core 数的 2–3 倍（如 120 core → 设 240~360），让每个 task 处理更少 key-group，减少内存峰值 - 当存在明显倾斜（如几个超级用户占了 70% 行数）：单纯调大分区数没用，得先加盐（salting）或用
approx_count_distinct预过滤，再设合理分区数（比如 120–180），否则倾斜 task 仍会拖垮整体 - 如果只是简单时间窗口（
partition by date_trunc('day', ts)），且天数不多（
最容易被忽略的一点：窗口函数的排序字段（orderBy）如果没建索引、也没提前过滤，会导致 shuffle 阶段读入大量无用数据参与排序 —— 此时调并行度只是治标。真正该做的是在 where 子句里收紧时间范围，或用 filter().withColumn(...).window(...) 把计算链路压得更紧凑。并行度解决的是“怎么分”，但前提是“分什么”已经足够干净。
