Spark分区优化实战：从输入切片到Shuffle重排的全链路调优

1. 为什么分区不是“加个参数就完事”，而是Spark性能的命门所在

你写完一个Spark作业，本地测试跑得飞快，一上集群就卡在Shuffle阶段，Stage卡住不动，Executor内存爆满，日志里反复刷出Failed to allocate X bytes——这时候翻文档、查Stack Overflow、问同事，十有八九会听到一句：“你分区没调好。”但没人告诉你，分区不是配置项，而是数据流经整个计算图时的“血管直径”和“血流节奏”。它直接决定数据如何被切片、如何被调度、如何被序列化、如何跨节点传输、如何被下游算子消费。一个不合理的分区数，会让本该并行执行的100个Task，实际只有12个在干活；会让本该均匀分布的1TB数据，90%挤在3个Executor里，剩下17个空转；更会让一次简单的groupByKey操作，从3分钟飙升到47分钟。

我最早在某高校实验室做图像特征聚合项目时就栽过这个跟头。原始数据是200万张图片提取的向量，存为Parquet，总大小约85GB。按常规思路，用spark.read.parquet("hdfs://...")加载后直接df.groupBy("category").agg(collect_list("feature"))，结果作业跑了2小时还没结束，YARN界面显示16个Executor中，5个GC时间占比超90%，3个OOM被Kill，其余全在Wait状态。后来逐层打点发现：read阶段生成了仅12个Partition，而集群有32核可用资源——相当于让32个人排队等12台机器干活，剩下20人干瞪眼。这不是代码写错了，是数据“路网规划”彻底失灵。

分区这件事，本质上是在回答三个关键问题：数据从哪来（Input Partitioning）、中间怎么分（Shuffle Partitioning）、结果往哪去（Output Partitioning）。这三个环节环环相扣，任何一个断点，都会让整条流水线淤塞。比如读HDFS文件时，分区数由minPartitions和文件块大小共同决定；做join或groupBy时，Shuffle前的partitionBy策略决定了网络传输量；最后写入Hive表时，repartition(200)可能比coalesce(200)多出3倍Shuffle数据量——这些都不是玄学，而是可计算、可验证、可优化的工程事实。

所以别再把spark.sql.shuffle.partitions=200当成万能膏药贴在spark-defaults.conf里就完事。真正懂分区的人，会在写第一行spark.read之前，就画出数据流动草图：源端文件有多少Block？目标并行度需要多少Task？中间Key的分布是否倾斜？输出目录是否支持并发写入？这就像修高速公路前必须先做地质勘测和车流模拟——分区，就是Spark应用的底层基建图纸。接下来，我们就一层层拆开这张图纸，看清楚每一条“车道”是怎么划出来的，为什么这么划，以及划错之后，你的作业到底会在哪个弯道翻车。

2. 分区设计的底层逻辑：从数据源头到计算终点的全链路拆解

2.1 输入分区：HDFS/Parquet/OSS上的“物理切片”如何映射为RDD分区

Spark读取外部存储时，输入分区数不是由spark.default.parallelism决定的，而是由数据源的物理布局和读取器的分片策略共同决定的。很多人误以为sc.textFile(path, minPartitions)里的minPartitions是强制指定分区数，其实它只是“最小建议值”，最终分区数仍受底层存储约束。

以HDFS为例：一个1.2GB的Parquet文件，若HDFS块大小为128MB，则该文件会被切分为10个Block（1.2GB ÷ 128MB ≈ 9.375 → 向上取整为10）。Spark的ParquetFileFormat读取器会为每个Block创建一个InputSplit，每个InputSplit对应一个Task，因此默认生成10个Partition。此时即使你传入minPartitions=50，Spark也不会强行切碎Block——因为HDFS不支持跨Block读取，强行切分会破坏数据局部性，导致大量远程读取。实测数据：当minPartitions=50但文件只有10个Block时，rdd.getNumPartitions返回的仍是10。

但如果你读的是多个小文件呢？比如1000个10MB的JSON文件，分布在HDFS同一目录下。此时spark.read.json("hdfs://path/to/jsons/")会为每个文件创建一个InputSplit（因JSON无块对齐概念），默认生成1000个Partition。这看似并行度高，实则灾难——每个Task处理10MB数据，启动开销（JVM初始化、序列化、网络握手）远超计算本身，CPU大量时间花在上下文切换上。我们曾在一个日志解析任务中观察到：1000个Partition的作业，总调度耗时占全程42%，而将小文件合并为100个100MB文件后，同样数据量，作业总耗时下降58%。

提示：判断输入分区是否合理，最直接的方法是执行df.rdd.getNumPartitions并对比集群核心数。理想比例是1.5~3倍（如32核集群，输入Partition数在48~96之间较优）。低于此值说明并行不足；高于3倍需警惕小文件病。

对于对象存储（如OSS、S3），情况更复杂。它们不提供类似HDFS的Block元信息，Spark依赖listStatus()获取文件列表后，按文件大小加权分片。例如读取10个文件（大小分别为1GB、1GB、500MB、500MB、200MB×6），Spark会计算总大小3.2GB，目标分区数设为100，则每个Partition目标大小≈32MB。算法会贪心合并小文件：前两个1GB文件各占31个Partition，500MB文件各占15个，剩余6个200MB文件被合并进其他Partition——最终生成的Partition数可能不是精确100，而是98或102。这种动态分片虽灵活，但也带来不确定性：同一份数据，不同时间点读取可能因文件列表顺序微变，导致Partition数浮动±5%。

2.2 Shuffle分区：groupByKey、join背后的“数据重排引擎”如何工作

当计算涉及Key-Value操作（如groupByKey、reduceByKey、join）时，Spark必须执行Shuffle——即按Key重新组织数据分布。此时的分区行为与输入分区完全解耦，由spark.sql.shuffle.partitions（SQL API）或spark.default.parallelism（RDD API）控制，但它的本质不是“分多少份”，而是“建多少个收件箱”。

想象一个快递分拣中心：所有包裹（数据）先被送到中心（Shuffle Write阶段），贴上目的地标签（HashPartitioner计算Key的hash值），然后按标签投入对应编号的格口（Partition）。下游Task（快递员）各自认领一个格口，取出所有包裹处理（Shuffle Read阶段）。spark.sql.shuffle.partitions=200的意思是：建200个格口，每个格口存放hash值模200等于其编号的数据。

关键点在于：格口数量固定，但每个格口的包裹量取决于Key的分布。如果Key严重倾斜（如90%数据Key为"UNKNOWN"），那么编号为hash("UNKNOWN") % 200的那个格口会塞满90%的包裹，而其他199个格口几乎为空。这就是典型的“数据倾斜”，它让200个Task中，1个Task要处理90%工作量，其余199个Task早早完成干等。我们曾处理用户行为日志，其中user_id="0"占全量数据的87%，groupByKey后单个Task耗时42分钟，其他Task平均23秒——整个Stage被拖垮。

更隐蔽的问题是Shuffle分区数与集群资源的错配。假设集群有32核，spark.sql.shuffle.partitions=200，意味着Shuffle Read阶段会启动200个Task并发读取。但YARN默认每个Executor只运行2个Task（spark.executor.cores=2），32核集群最多同时运行16个Executor × 2 = 32个Task。结果就是：200个Task被排队调度，前32个Task跑完后，系统才释放资源启动下一批——Task启动延迟高达15~20秒，整体吞吐量暴跌。实测数据：将spark.sql.shuffle.partitions从200降至64（32核×2），同作业Shuffle阶段耗时从18分钟降至6分钟，GC次数减少73%。

2.3 输出分区：repartition与coalesce的“分流”与“合流”本质差异

写入结果时，常需调整分区数以匹配下游系统要求（如Hive表要求200个文件）或避免小文件（如合并为5个大文件）。此时repartition(n)和coalesce(n)常被混用，但二者原理天壤之别：

• repartition(n)：强制触发全量Shuffle。无论当前多少分区，都新建n个Partition，通过HashPartitioner重分布所有数据。适合需要均匀重分布的场景（如join后按新Key重新分区）。
• coalesce(n)：不触发Shuffle的窄依赖合并。它仅将相邻Partition合并（如原100分区→coalesce(20)，则Partition 0+1+2+3+4→新Partition 0，以此类推），要求n小于当前分区数。适合减少分区数且不关心数据分布的场景（如写入前合并小文件）。

陷阱在于：coalesce(200)对100分区的RDD无效（参数n必须≤当前分区数），而repartition(200)对100分区RDD会触发Shuffle——即使你只是想增加分区数。我们曾为提升写入Hive并发度，对100分区的DataFrame执行repartition(200)，结果Shuffle Write数据量暴增2.3倍（因原100分区数据被随机打散到200分区），网络IO成为瓶颈，写入耗时反增40%。

正确做法是：先评估是否真需Shuffle。若源数据Key分布已较均匀（如user_id哈希后分散），可直接用repartition(200)；若Key有倾斜，应先salting（加随机前缀）再repartition；若只是避免小文件且不介意数据分布，用coalesce(20)比repartition(20)快3倍以上。

3. 核心实操：分区数的量化计算与动态调优四步法

3.1 第一步：基线测量——用三行代码摸清当前分区健康度

不要凭经验猜，用数据说话。在Spark Shell或Notebook中执行以下诊断脚本：

# 获取当前DataFrame的分区信息 df_explain = df.explain(mode="extended") print("=== 物理计划摘要 ===") # 手动提取分区数（实际使用中需解析explain输出） # 更直接的方法： print(f"当前分区数: {df.rdd.getNumPartitions()}") print(f"数据量估算: {df.count()} 行 (注意：count()是Action，慎用于大数据集)") # 对于大表，用采样估算 sample_df = df.sample(False, 0.01) # 采样1% approx_count = sample_df.count() * 100 print(f"采样估算总行数: {approx_count:,}") # 检查分区大小分布（需启用AQE或手动统计） from pyspark.sql.functions import input_file_name, size, col if "input_file_name" in df.columns: file_size_df = df.withColumn("file", input_file_name()).groupBy("file").count() file_size_df.agg( {"count": "min", "count": "max", "count": "avg"} ).show()

重点看三个指标：

• 分区数 vs 集群核心数比值：理想区间1.5~3。比值<1.2说明并行不足；>5需警惕小文件或过度切分。
• 分区数据量标准差：用df.rdd.mapPartitions(lambda it: [sum(1 for _ in it)]).collect()获取各Partition行数，计算标准差。若标准差 > 均值的50%，存在明显数据倾斜。
• Shuffle阶段Task耗时方差：在Spark UI的Stage页面，看Tasks列表的Duration列。若最长Task耗时 > 平均耗时的3倍，基本可判定倾斜。

我们曾诊断一个ETL任务，发现getNumPartitions()返回48，集群32核，比值1.5属合理范围。但进一步检查各Partition行数，发现48个值中，45个在80万~120万之间，3个为0（因上游过滤导致空Partition）。这3个空Partition虽不耗时，却占用调度资源——后续通过filter(_.nonEmpty)清理，作业启动延迟降低18%。

3.2 第二步：输入分区优化——针对不同数据源的定制化方案

HDFS/Parquet文件：用maxPartitionBytes精准控件大小

Spark 3.0+引入spark.sql.files.maxPartitionBytes（默认128MB），它比minPartitions更智能：按文件总大小 ÷ 目标分区大小，动态计算最优分区数，并尽量保持每个Partition接近该大小。

例如读取总大小为2.5GB的Parquet目录：

• 默认maxPartitionBytes=128MB→ 目标分区数 ≈ 2.5GB ÷ 128MB ≈ 20 → 实际生成20个Partition
• 设为256MB→ 目标分区数 ≈ 10 → 生成10个Partition
• 设为64MB→ 目标分区数 ≈ 40 → 生成40个Partition

关键优势：它会智能合并小文件。若目录含100个10MB文件，maxPartitionBytes=128MB会将8~10个文件合并为一个Partition，避免100个超小Partition。

注意：maxPartitionBytes仅对Parquet/ORC等支持分片的格式生效，对Text/JSON无效。

小文件治理：合并而非跳过

面对海量小文件（如Kafka实时写入的每分钟1000个JSON），spark.sql.files.ignoreCorruptFiles=true只能跳过损坏文件，无法解决性能问题。必须前置合并：

# 方案1：HDFS层面合并（推荐） # 使用Hadoop distcp合并小文件到新路径 # hadoop distcp -update -m 10 hdfs://source/path hdfs://dest/merged_path # 然后读取merged_path # 方案2：Spark内合并（适用于无法改源场景） from pyspark.sql import SparkSession spark = SparkSession.builder.appName("MergeSmallFiles").getOrCreate() # 读取所有小文件 small_files_df = spark.read.json("hdfs://path/to/small/files/*") # 写入新路径，指定分区数控制文件数 small_files_df.repartition(50).write.mode("overwrite").json("hdfs://path/to/merged/")

实测：1000个10MB JSON文件，直接读取耗时8.2分钟；先repartition(50)合并为50个200MB文件再读取，耗时降至3.1分钟，且后续计算更稳定。

对象存储（OSS/S3）：用recursiveFileLookup规避目录遍历瓶颈

读取嵌套目录（如oss://bucket/logs/year=2023/month=01/day=01/）时，Spark默认递归列出所有子目录，当目录层级深、文件数多时，listStatus()耗时可达分钟级。开启recursiveFileLookup可跳过中间目录：

spark.read.option("recursiveFileLookup", "true") \ .parquet("oss://bucket/logs/year=2023/")

此选项让Spark直接扫描year=2023/下所有Parquet文件（忽略month/day子目录），将listStatus耗时从210秒降至8秒。配合maxPartitionBytes=256MB，可快速获得合理分区数。

3.3 第三步：Shuffle分区调优——从静态配置到动态适配

静态调优：基于数据量的黄金公式

spark.sql.shuffle.partitions不应设为固定200，而应按集群总内存 × 0.8 ÷ 单Partition处理内存估算。单Partition处理内存经验值为200MB（兼顾序列化开销和GC压力）：

目标Shuffle分区数 = (集群Executor总内存 × 0.8) ÷ 200MB

例如：8台Executor，每台32GB内存 → 总内存256GB → 目标分区数 = (256 × 0.8) ÷ 0.2 ≈ 1024。但需结合核心数约束：32核集群，最大并发Task数为32（若spark.executor.cores=1）或64（若spark.executor.cores=2），因此最终取min(1024, 64) = 64。

我们在线上环境验证该公式：将shuffle.partitions从200调至64后，Shuffle Write峰值内存下降37%，GC时间减少61%，Stage失败率归零。

动态调优：AQE（Adaptive Query Execution）实战

Spark 3.0+的AQE可自动优化Shuffle分区。启用后，Spark在Shuffle Write完成后，根据实际数据量分布，动态合并小Partition、拆分大Partition、优化Join策略。

启用方式（无需改代码）：

# spark-defaults.conf spark.sql.adaptive.enabled true spark.sql.adaptive.coalescePartitions.enabled true spark.sql.adaptive.skewJoin.enabled true

效果实测：一个含join的ETL作业，AQE启用前，因数据倾斜导致1个Task耗时15分钟；启用后，AQE检测到倾斜，将大Partition拆分为5个子Partition，由5个Task并行处理，最长Task耗时降至2.3分钟，整体作业提速5.2倍。

注意：AQE需配合spark.sql.adaptive.localShuffleReader.enabled true（本地读取优化）和spark.sql.adaptive.skewJoin.enabled true（倾斜Join）才能发挥最大效用。

3.4 第四步：输出分区控制——写入Hive/Parquet的精准文件数管理

Hive表写入：用bucketBy替代盲目repartition

Hive表常需固定文件数（如分区表要求每个分区200个文件）。若用df.repartition(200).write.saveAsTable("db.table")，会触发Shuffle且无法保证文件数（因HiveWriter内部可能再分片）。正确姿势是bucketBy：

df.write \ .mode("overwrite") \ .bucketBy(200, "user_id") \ # 按user_id哈希分200桶 .sortBy("event_time") \ # 每桶内按时间排序 .saveAsTable("db.table")

bucketBy在Shuffle Write阶段直接按user_id哈希写入200个文件（无额外Shuffle），且文件名带bucket_000XX标识，便于运维定位。实测：对10亿行数据，bucketBy(200)写入耗时比repartition(200)快2.8倍，网络IO减少64%。

Parquet写入：maxRecordsPerFile控制单文件行数

当需限制单文件大小（如避免HDFS小文件），maxRecordsPerFile比repartition更精准：

df.write \ .option("maxRecordsPerFile", "5000000") \ # 每文件500万行 .mode("overwrite") \ .parquet("hdfs://path/to/output/")

Spark Writer会动态监控当前文件行数，达到阈值即切新文件。相比repartition(n)的粗粒度控制，它能生成大小更均匀的文件。我们处理用户画像数据时，用maxRecordsPerFile=5000000生成的Parquet文件，95%在800MB~1.2GB之间（HDFS块128MB），完美匹配HDFS局部性。

4. 常见问题排查与避坑指南：那些年我们踩过的分区深坑

4.1 问题速查表：症状、根因与一键修复

4.2 独家避坑技巧：来自生产环境的血泪总结

技巧1：用repartitionByRange替代repartition应对有序数据

当数据已按某字段排序（如event_time），用repartition(200)会打乱顺序，导致后续orderBy需二次Shuffle。而repartitionByRange(200, "event_time")按时间范围切分，既保证200个Partition，又维持各Partition内数据有序。我们处理时序日志时，用此方法将orderBy("event_time").limit(100)耗时从4.2分钟降至23秒。

技巧2：coalesce后立即cache()防重复计算

coalesce是窄依赖，不触发Job，但若后续有多次Action（如count()和show()），每次都会重算。务必在coalesce后加cache()：

# 错误：每次Action都重算coalesce df.coalesce(10).count() df.coalesce(10).show() # 正确：计算一次，多次复用 df_coalesced = df.coalesce(10).cache() # cache()是Action，触发计算 df_coalesced.count() df_coalesced.show()

技巧3：自定义Partitioner避开Hash冲突

默认HashPartitioner对字符串Key用String.hashCode()，但Java的hashCode()在长字符串下易冲突。我们曾遇到user_id为UUID字符串，hashCode()后模200，90%落入同一Partition。解决方案：自定义Partitioner用Murmur3哈希：

import org.apache.spark.util.collection.BitSet import java.nio.ByteBuffer import com.google.common.hash.Hashing class Murmur3Partitioner(numParts: Int) extends Partitioner { override def numPartitions: Int = numParts override def getPartition(key: Any): Int = { val hash = Hashing.murmur3_32().hashString(key.toString, charset).asInt() math.abs(hash) % numParts } } // 使用：rdd.partitionBy(new Murmur3Partitioner(200))

实测：UUID Key的分布标准差从1200降至45，各Partition数据量差异<5%。

技巧4：AQE失效时的手动adaptive回退

AQE在某些场景会失效（如自定义UDF阻断优化）。此时可手动模拟AQE的Partition合并：

# 获取各Partition行数 partition_counts = df.rdd.mapPartitions(lambda it: [sum(1 for _ in it)]).collect() # 计算平均值，合并小于平均值50%的Partition avg_count = sum(partition_counts) / len(partition_counts) threshold = avg_count * 0.5 # 用coalesce合并小Partition（需自定义逻辑，此处略）

虽然繁琐，但在关键任务中，这行代码曾帮我们避免了一次线上事故。

5. 进阶实践：分区与Spark核心机制的深度耦合

5.1 分区与内存管理：为何spark.memory.fraction必须随分区数动态调整

Spark内存模型中，spark.memory.fraction（默认0.6）定义了堆内存中Execution + Storage共享的比例。但每个Partition的Shuffle缓冲区（spark.shuffle.file.buffer，默认32KB）是独立分配的。若分区数从200增至2000，Shuffle缓冲区总内存需求从6.4MB飙升至64MB——这部分内存来自spark.memory.fraction分配的区域。

当分区数激增而spark.memory.fraction未调高时，Shuffle缓冲区争抢Storage内存，导致缓存数据被频繁驱逐，Cache Hit Rate暴跌。我们曾将shuffle.partitions从200调至1000，未调整内存参数，结果Storage Memory使用率从40%升至95%，BlockManager日志频繁出现Dropping block警告，作业失败率上升至35%。

解决方案：分区数每增加5倍，spark.memory.fraction至少提升0.1。例如原配置fraction=0.6，分区数从200→1000，则设为0.7；若再增至2000，需设为0.8。同时调高spark.shuffle.file.buffer至64KB，减少缓冲区数量。

5.2 分区与序列化：Kryo注册类为何要按分区粒度优化

Spark序列化框架（Kryo）的注册类列表，在Driver端序列化后广播到所有Executor。但每个Partition的Task在反序列化时，会加载完整注册表。若注册了1000个类，而实际每个Partition只用到其中50个，其余950个类的反序列化开销被白白浪费。

优化思路：按业务模块拆分注册表，让每个Job只注册其用到的类。例如用户行为分析Job只注册Event、UserProfile类；商品推荐Job只注册ItemFeature、CollabFilterResult类。我们实施此方案后，Task启动时间从1.8秒降至0.4秒，原因正是减少了Kryo注册表的反序列化负载。

5.3 分区与容错：为何spark.speculation对高分区数作业更有效

spark.speculation（推测执行）机制会监控Task进度，对明显慢于平均的Task，在其他节点启动副本。其触发阈值spark.speculation.interval（默认100ms）和spark.speculation.multiplier（默认1.5）在高分区数作业中更易生效——因为Task数量多，统计样本充足，慢Task识别更精准。

但需注意：speculation会增加集群负载。我们的经验是：分区数>200时开启speculation，分区数<100时关闭。开启后，将spark.speculation.quantile从默认0.75调至0.9，避免过早启动副本（因少量Task慢属正常波动）。

5.4 分区与资源调度：YARN队列选择如何影响分区效率

YARN的Capacity Scheduler中，不同队列的maximum-allocation-mb（单Container最大内存）直接影响Executor内存上限。若队列设置maximum-allocation-mb=8192（8GB），而你申请--executor-memory 12g，YARN会强制降为8GB，导致Executor内存不足，频繁GC。

此时，即使分区数合理，也会因内存瓶颈表现异常。解决方案：为高分区数作业（>200）申请专用大内存队列，或调整分区数匹配队列能力。例如队列最大8GB，则单Executor最多承载8GB ÷ 200MB ≈ 40个Partition，故shuffle.partitions应设为集群Executor数 × 40。

我在某公司部署实时风控系统时，就因未注意队列内存限制，将shuffle.partitions设为400，结果Executor频繁OOM。切换至max-memory=16g的队列后，问题迎刃而解。这提醒我们：分区优化不能脱离基础设施谈。

6. 实战案例复盘：从日志解析到实时报表的全链路分区治理

6.1 项目背景：千万级设备日志的实时清洗与聚合

某物联网平台每日产生1200万设备上报日志，格式为JSON，包含device_id、timestamp、sensor_data（嵌套结构）等字段。原始数据由Flume写入HDFS/logs/raw/year=2023/month=06/day=15/，单日文件数约3200个（大小10MB~50MB不等）。业务需求：

• 实时清洗（过滤无效日志、解析sensor_data）
• 每10分钟窗口聚合device_id的平均温度、最高湿度
• 结果写入Hive表，供BI工具查询

初始方案（失败）：

# 读取 df = spark.read.json("hdfs://logs/raw/year=2023/month=06/day=15/") # 清洗 clean_df = df.filter("device_id IS NOT NULL AND sensor_data.temp IS NOT NULL") # 聚合 result_df = clean_df.groupBy("device_id").agg( avg("sensor_data.temp").alias("avg_temp"), max("sensor_data.humidity").alias("max_humidity") ) # 写入 result_df.write.mode("append").saveAsTable("iot.device_summary")

结果：作业平均耗时28分钟，失败率22%，YARN日志显示大量Container killed by YARN for exceeding memory limits。

6.2 分区问题诊断与根因定位

第一步，运行诊断脚本：

print(f"输入分区数: {df.rdd.getNumPartitions()}") # 输出: 3200 print(f"清洗后行数: {clean_df.count()}") # 输出: 1120万 # 检查分区大小 sizes = clean_df.rdd.mapPartitions(lambda it: [sum(1 for _ in it)]).collect() print(f"分区行数标准差: {np.std(sizes):.0f}") # 输出: 12500（均值3500，标准差远超均值）

发现：3200个超小Partition（因3200个小文件），且数据倾斜（device_id为"UNKNOWN"的日志占15%）。

第二步，查看Shuffle阶段：

• groupByKey触发Shuffle，shuffle.partitions=200（默认）
• 但3200个输入Partition → 200个Shuffle Partition，每个Shuffle Partition需合并16个输入Partition的数据，内存压力巨大。

6.3 分区优化方案与效果对比

优化步骤：

1. 输入层合并：用distcp将3200个文件合并为64个500MB文件
2. 读取层控参：spark.conf.set("spark.sql.files.maxPartitionBytes", "500MB")→ 输入分区数=64
3. 倾斜处理：对device_id="UNKNOWN"加盐

   from pyspark.sql.functions import when, rand, concat salted_df = clean_df.withColumn( "salted_device_id", when(col("device_id") == "UNKNOWN", concat(lit("UNKNOWN_"), (rand() * 100).cast("int"))) .otherwise(col("device_id")) )

4. Shuffle层调优：spark.conf.set("spark.sql.shuffle.partitions", "128")（64核集群×2）
5. 输出层控制：bucketBy(128, "salted_device_id")写入Hive

效果对比：