如何优化PySpark subtract性能？1TB级DataFrame运行缓慢求解

针对两个1TB级别的DataFrame执行subtract操作速度慢的问题，可以从数据预处理、执行逻辑、资源配置等多个维度优化：

• 精简数据范围

  • 只保留必要列：差集计算仅需用于判断行唯一性的字段，提前过滤无关列，减少数据传输和计算量。示例：

    # 假设仅需col1、col2作为匹配键
    df1 = df1.select("col1", "col2")
    df2 = df2.select("col1", "col2")
    

  • 提前去重：如果df1或df2存在重复行，先执行distinct()减少后续计算的行数：

    df1 = df1.distinct()
    df2 = df2.distinct()
    

• 优化差集计算逻辑

  • 用Left Anti Join替代subtract：subtract底层实现差集，但Left Anti Join的优化规则更灵活，能更好利用Spark的执行计划优化。示例：

    # 按匹配键执行左反连接，等价于df1.subtract(df2)
    diffDF = df1.join(df2, on=["col1", "col2"], how="left_anti")
    

  • 广播小表：如果其中一个DataFrame（比如df2）体积远小于另一个，广播小表避免全量Shuffle：

    from pyspark.sql.functions import broadcast
    diffDF = df1.subtract(broadcast(df2))
    

• 调整数据分区与Shuffle配置

  • 合理设置分区数：根据集群资源调整DataFrame的分区数，避免分区过大（单分区数据量超过Executor内存）或过小（过多小分区导致调度开销）。示例：

    # 根据集群核心数设置分区数，比如200个分区
    df1 = df1.repartition(200)
    df2 = df2.repartition(200)
    

  • 开启Shuffle优化：在Spark配置中开启外部Shuffle服务和Shuffle压缩，减少IO开销：

    spark.shuffle.service.enabled=true
    spark.shuffle.compress=true
    

• 优化存储格式与数据布局

  • 使用列式存储格式：将数据转换为Parquet或ORC格式，这类格式支持列式读取、 predicate pushdown，大幅降低IO读取量。
  • 分区/分桶存储：如果数据有天然维度（如时间、地域），提前对DataFrame进行分区或分桶，让差集计算仅扫描相关分区的数据，减少扫描范围。

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