超2万列Spark去重值批量处理优化求助

我之前也踩过大规模列去重统计的坑，你的思路本身没问题，但循环单列处理完全浪费了Spark的分布式并行优势——毕竟Spark是为批量数据处理设计的，单列循环相当于把它当单机工具用了，慢是必然的。下面分享几个亲测有效的优化方案：

先别循环单列跑approxCountDistinct，直接把所有列一次性丢进agg里批量计算，Spark会自动把任务并行分发到集群节点，效率能提升一个数量级：

from pyspark.sql import functions as F

# 一次性计算所有列的近似去重数，用*展开所有列的聚合表达式
approx_distinct_df = df.agg(*[F.approxCountDistinct(col).alias(col) for col in df.columns])

# 把结果转成字典，方便后续筛选需要提取具体值的列
approx_distinct_map = approx_distinct_df.collect()[0].asDict()

# 筛选出近似计数低于阈值的列
target_columns = [col for col, cnt in approx_distinct_map.items() if cnt < 10]

这一步同样要避免单列循环，推荐两种思路，按需选择：

思路A：宽表转长表，分布式去重后分组收集（适合大量目标列） #

把所有目标列转成「列名-列值」的键值对长表，一次性完成去重和分组收集，全程利用Spark的分布式能力，只需要两次集群作业：

from pyspark.sql import functions as F

if target_columns:
    # 构造每个列的(列名, 列值)结构体
    column_structs = [F.struct(F.lit(col).alias("col_name"), F.col(col).alias("col_value")) 
                      for col in target_columns]
    
    # 把所有结构体合并成数组，炸开成行后去重
    long_format_df = df.select(F.explode(F.array(*column_structs)).alias("data")) \
                       .select("data.col_name", "data.col_value") \
                       .distinct()
    
    # 按列名分组，收集所有去重值
    distinct_result_df = long_format_df.groupBy("col_name") \
                                       .agg(F.collect_list("col_value").alias("distinct_values"))
    
    # 转成字典得到最终结果
    distinct_values_map = {row.col_name: row.distinct_values for row in distinct_result_df.collect()}

思路B：批量Select+Distinct，本地拆解结果（适合少量目标列） #

如果目标列数量不多（比如几百列以内），可以直接一次性Select所有目标列后去重，再在本地把每行数据拆解到对应列的列表里，代码更简洁：

if target_columns:
    # 一次性获取所有目标列的去重行
    distinct_rows = df.select(target_columns).distinct().collect()
    
    # 初始化结果字典
    distinct_values_map = {col: [] for col in target_columns}
    
    # 遍历去重后的行，把值添加到对应列的列表中
    for row in distinct_rows:
        for col in target_columns:
            val = row[col]
            if val not in distinct_values_map[col]:
                distinct_values_map[col].append(val)

• 调整分区数：如果DataFrame分区太少，集群节点闲得慌，先执行df = df.repartition(集群CPU核心数*2~3)，让每个节点都有足够任务可跑。
• 分组处理超大量列：如果要处理上万列，把列分成若干组（比如每500列一组），分组执行近似计数和去重提取，避免单次作业生成过多小任务，增加Spark调度开销。
• 提前过滤冗余数据：如果原始数据有大量空值或无效行，先过滤掉再处理，比如df = df.filter(F.coalesce(*[F.col(c).isNotNull() for c in target_columns]))，减少后续处理的数据量。

内容的提问来源于stack exchange，提问作者breakingduck