PySpark并行读取多文件至独立DataFrame遇Pickle错误，求解决方法

为什么会出现Pickle错误？ #

Spark DataFrame本质是分布式逻辑执行计划，并非本地内存中的数据结构，无法通过Python的Pickle序列化机制在进程/线程间传递。你用ThreadPool的starmap时，会尝试序列化函数返回的DataFrame，这直接触发了序列化失败的错误。

推荐解决方案（利用Spark原生并行能力） #

Spark本身就是为分布式大数据处理设计的，不需要手动用线程池实现并行读取，以下是两种可行方案：

方案1：直接循环读取（懒加载天然并行） #

Spark的read.csv是懒执行操作，只有当你触发count()、show()等action时才会实际读取数据。直接用列表推导式循环读取多个文件，Spark调度器会自动优化为并行执行：

file_paths = [file1, file2, file3, ...]
df_list = [spark.read.csv(path) for path in file_paths]

每个DataFrame都是独立的逻辑执行计划，后续可分别对它们执行action操作，Spark会并行处理读取任务。

方案2：用Spark RDD实现显式并行 #

如果需要更精细的并行控制，可以将文件路径转为RDD，通过map操作分发读取任务到各个executor：

file_paths = [file1, file2, file3, ...]
# 并行化文件路径，指定分区数控制并行度
path_rdd = spark.sparkContext.parallelize(file_paths, numSlices=10)

def read_single_csv(path):
    return spark.read.csv(path)

# 收集所有独立DataFrame的引用（仍为逻辑计划，未实际读取）
df_list = path_rdd.map(read_single_csv).collect()

注意：collect()只是把DataFrame的逻辑计划拉回driver，实际数据读取仍在executor上分布式执行，避免内存溢出。

不推荐的替代方案（本地多线程读取） #

如果非要用本地线程池，只能返回pandas DataFrame，但10-15GB的大文件极易导致内存溢出，仅适合小文件场景：

from multiprocessing.pool import ThreadPool
import pandas as pd

def read_local_csv(file_path):
    # 大文件需添加chunksize参数分块读取，避免OOM
    return pd.read_csv(file_path, chunksize=100000)

pool = ThreadPool(10)
# starmap要求每个任务参数是元组，需将单个路径转为单元素元组
df_iterators = pool.starmap(read_local_csv, [(path,) for path in file_paths])

此方案仅适合内存足够容纳单份文件的场景，否则需处理分块迭代器。

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