Pyspark中approx_count_distinct函数循环执行卡顿/性能极差问题排查

核心原因：Spark逐行循环是典型反模式 #

你的代码在数据量从1万涨到10万时性能暴跌，本质是违背了Spark的设计初衷：

• Driver-Executor通信过载：逐行从Spark DataFrame取数（比如collect()、toLocalIterator()）会把分布式存储的数据集拉到Driver节点，每一行的读取、处理都要跨节点通信，10万次循环的通信开销呈指数级增长。
• DataLineage无限膨胀：每次循环用union()合并新行时，Spark会生成新的执行计划，且保留完整的依赖链（lineage）。随着循环次数增加，lineage会越来越长，Spark优化器解析和执行计划的成本急剧上升，甚至可能触发栈溢出。
• O(n²)级别的重复计算：每次循环都要对整个HLL DataFrame调用approx_count_distinct，随着HLL DataFrame行数增长，单次计算耗时线性增加，10万次循环的总复杂度直接变成O(n²)，这是导致15小时还没跑完的核心原因。

针对性解决方案 #

方案1：用批量处理重构逻辑（推荐） #

Spark的优势是批量分布式运算，完全不需要逐行循环。你的逻辑是找出加入后不会改变approx_count_distinct估算值的行，可以通过以下方式实现：

from pyspark.sql import functions as F

# 计算全量数据的基准估算值
base_estimate = df.select(F.approx_count_distinct("id")).first()[0]

# 批量识别对估算值无贡献的行（逻辑可根据实际场景调整）
y_df = df.withColumn("temp_id", F.concat(F.col("id"), F.lit(""))) \
         .groupBy("id") \
         .agg(F.approx_count_distinct("temp_id").alias("curr_estimate")) \
         .filter(F.col("curr_estimate") == base_estimate) \
         .drop("curr_estimate")

方案2：用本地HLL库替代Spark的approx_count_distinct（数据量允许时） #

如果你的数据集可以完整拉到Driver节点内存（10万条数据完全没问题），可以用Python本地的HLL库（比如hyperloglog）处理，避免Spark的分布式开销：

import hyperloglog
from pyspark.sql import Row

# 把数据拉到本地内存
local_rows = df.collect()

# 初始化本地HLL实例（误差率和Spark的approx_count_distinct对齐）
hll = hyperloglog.HyperLogLog(0.01)
y_rows = []

for row in local_rows:
    prev_count = len(hll)
    hll.add(row.id)
    curr_count = len(hll)
    if prev_count == curr_count:
        y_rows.append(row)

# 将结果转回Spark DataFrame
y_df = spark.createDataFrame(y_rows)

方案3：用RDD分区处理减少Driver交互 #

如果数据量太大无法拉到本地，可在Executor端分区处理，用累加器跟踪HLL状态：

import hyperloglog
from pyspark import AccumulatorParam

# 自定义HLL累加器
class HLLAccumulatorParam(AccumulatorParam):
    def zero(self, initial_value):
        return hyperloglog.HyperLogLog(0.01)
    
    def addInPlace(self, accumulator, value):
        accumulator.add(value)
        return accumulator

# 注册累加器
hll_accum = spark.sparkContext.accumulator(hyperloglog.HyperLogLog(0.01), HLLAccumulatorParam())

# 分区处理，找出符合条件的行
def process_partition(iterator):
    local_hll = hyperloglog.HyperLogLog(0.01)
    local_hll.merge(hll_accum.value)
    result = []
    for row in iterator:
        prev_count = len(local_hll)
        local_hll.add(row.id)
        curr_count = len(local_hll)
        if prev_count == curr_count:
            result.append(row)
        hll_accum.add(row.id)
    return result

y_rdd = df.rdd.mapPartitions(process_partition)
y_df = y_rdd.toDF()

关键总结 #

Spark是为批量分布式计算设计的，逐行循环是绝对的反模式——小数据量（1万）时可能看不出问题，但数据量放大后，通信开销、lineage膨胀、重复计算会直接导致性能雪崩。必须重构逻辑，利用Spark的批量操作或本地处理（数据量允许时）来规避这些问题。

内容的提问来源于stack exchange，提问作者Beatriz Correa Castillo