Spark中合并DataFrame时的最佳缓存位置及性能优化问询

我正读取多个单文件大小为200GB、包含数十亿行数据的文件，需对每个文件进行处理、合并后再做二次处理，现需明确Spark中DataFrame的最佳缓存位置。我的代码如下：

def load_and_process_single_file(path):
    df = spark.read.parquet(path) # This has 1B+ rows
   
    # Some transformations -> I guess there's no need caching these?
    df = df.withColumn('a', something)
    df = df.select(...)
    df = df.where(...)

    # Calculate counts
    df = df.groupby('a').count() # This has just ~10K rows

    return df

for i in range(n):
    df_single = load_and_process_single_file(path[i])

    if(i == 0):
        df_all = df_single 
    else:
        df_all = df_all.join(df_single, df_all.a == df_single.a, 'full_outer')
        # More transformations
        df_all = df_all.withColumn(...)
        df_all = df_all.fillna(0).withColumn(...)
        df_all = df_all.select(..)

        # Does uncommenting the below line change if something should be cached?
        # df_all.show()

    # Save df_all to disk after each iteration
    df_all.write.csv(...)

请问：

1. load_and_process_single_file函数内是否确实无Action操作，因此无需进行缓存？
2. 哪些位置是调用.cache()的明显合适点？
3. 尝试处理7个文件（n=7），耗时30小时仍未完成，除缓存外还有哪些性能优化手段可用？

问题1：load_and_process_single_file函数内是否无需缓存？ #

是的，这个函数里全是转换操作（Transformation），没有触发计算的行动操作（Action），Spark此时仅构建逻辑执行计划，不会真正执行计算。而且函数最终返回的是经过groupby后的小数据集（仅约1万行），后续即使触发Action，计算量也极小，所以函数内确实不需要缓存。

问题2：适合调用.cache()的位置 #

1. 函数返回的df_single：单个大文件处理后得到的df_single是小数据集，缓存它可避免后续join时重复执行整个文件的读取、转换逻辑。可以在函数末尾添加return df_single.cache()，或者调用后立即缓存：df_single = load_and_process_single_file(path[i]).cache()。

2. 迭代中的df_all：每次join后的df_all数据量会逐步增加，但因按a聚合，量级仍可控，缓存它可避免下一次迭代时重复执行之前所有的join和转换逻辑。建议在else分支的转换完成后，执行df_all = df_all.cache()。

3. 关于df_all.show()：如果开启这个Action，Spark会触发计算；若df_all未缓存，后续的write.csv会重新执行全部逻辑。因此如果有show操作，必须先缓存df_all，避免重复计算。

问题3：除缓存外的性能优化手段 #

• 优化存储格式：

  • 输出放弃CSV，改用Parquet或ORC这类列式存储格式，这类格式读写速度更快、支持压缩，能大幅降低磁盘IO开销（CSV作为文本格式，大数据量下读写效率极低）。
  • 读取Parquet时，确认spark.sql.parquet.filterPushdown参数已开启（默认开启），确保谓词下推生效，减少不必要的数据读取。

• 调整集群资源与Shuffle参数：

  • 增加Executor的内存和CPU核数，比如设置--executor-memory 32G --executor-cores 8（根据集群总资源调整），为大文件的Shuffle操作（groupby、join）提供足够资源。
  • 增大spark.sql.shuffle.partitions参数（默认200），可设置为1000-2000，避免单个Shuffle分区过大导致内存溢出或处理缓慢；开启spark.shuffle.service.enabled优化Shuffle数据拉取效率。

• 优化join逻辑：

  • 利用广播连接（Broadcast Join）：因为df_single是小数据集，可在join前执行df_single = df_single.broadcast()，或设置spark.sql.autoBroadcastJoinThreshold（比如设为100MB）让Spark自动启用广播，避免Shuffle操作。
  • 替换迭代式join：先收集所有df_single做union，再统一groupby('a')求和，union是窄依赖无需Shuffle，比多次full outer join效率高得多。

• 减少重复计算：

  • 若不缓存df_all，可将每次迭代后的df_all写入Parquet文件，下一次迭代直接读取该文件，而非重新计算之前的所有逻辑。

• 精简转换步骤：

  • 合并多个withColumn操作，减少转换环节；优先执行where过滤再做列转换，提前缩小数据处理量级。

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