问题现象与本质分析
当使用PySpark的distinct()方法处理大规模数据集时，内存溢出(OutOfMemoryError)是最常见的致命错误。典型报错表现为java.lang.OutOfMemoryError: Java heap space，其本质是Spark的哈希去重机制需要在内存中维护所有唯一键的哈希表。当数据集基数(cardinality)过高时：
- 执行器(Executor)的JVM堆空间无法容纳全部键值
- Shuffle阶段产生数据倾斜
- 序列化/反序列化开销指数增长
7种核心解决方案
1. 分区动态调整策略
# 通过采样预估数据分布
sample_ratio = min(1.0, 1000000 / df.count())
df.repartition(int(df.count()*sample_ratio), "key_column").distinct()采用自适应分区(Adaptive Partitioning)技术，根据数据基数动态调整分区数，避免固定分区导致的负载不均。
2. 内存优化配置矩阵
| 参数 | 推荐值 | 作用域 |
|---|---|---|
| spark.executor.memoryOverhead | 2GB+ | YARN/K8s环境 |
| spark.sql.shuffle.partitions | 200-1000 | 全局配置 |
| spark.default.parallelism | cores*2-4 | RDD操作 |
3. 替代方案性能对比
针对不同场景可选择替代方案：
- 近似去重：
approx_count_distinct(误差<5%) - 分桶处理：先按哈希分桶再各桶内去重
- 持久化+迭代：分批处理并合并结果
3种监控诊断方法
1. 内存压力指标监控
通过Spark UI观察：
- GC时间占比 >30%需警惕
- Storage Memory峰值波动
- Shuffle Read/Write不对称
2. 序列化效率分析
使用spark.driver.extraJavaOptions=-XX:+HeapDumpOnOutOfMemoryError获取堆转储文件，用Eclipse MAT分析对象占用情况，重点关注：
- UnsafeRow对象数量
- ByteBuffer分配情况
- HashSet扩容频率
深度优化案例
某电商平台处理4.2亿用户行为记录的去重时，通过组合方案解决OOM：
df.withColumn("bucket", F.hash("user_id") % 1000)
.repartition(1000, "bucket")
.persist(StorageLevel.MEMORY_AND_DISK)
.groupBy("bucket")
.agg(F.collect_set("user_id").alias("distinct_users"))关键技术点：
- 两层分治策略：哈希分桶+分区处理
- 混合持久化：内存+磁盘联合存储
- 聚合转换：distinct→collect_set转换