如何用sc.parallelize()并行读取S3上的100个Parquet文件构建PySpark数据框

你已经有了S3文件路径列表，用sc.parallelize()实现并行读取的核心思路是把路径分发给各个Executor，让它们同时处理对应文件。下面是具体的实现方案：

代码实现（PySpark为例） #

假设你已经生成了完整的S3文件路径列表file_paths，且SparkContext（sc）和SparkSession（spark）已初始化：

# 示例：你的S3文件路径列表
file_paths = ["s3://your-bucket/dir/file1.parquet", "s3://your-bucket/dir/file2.parquet", ...]

# 将路径列表转为RDD，指定分区数（推荐和文件数匹配，或设为集群核心数的2-3倍）
paths_rdd = sc.parallelize(file_paths, numSlices=100)

# 定义分区级读取函数，批量处理每个分区内的文件
def read_parquet_in_partition(paths):
    from pyspark.sql import SparkSession
    # 获取当前活跃的SparkSession
    spark = SparkSession.getActiveSession()
    if not paths:
        return []
    # 批量读取分区内的所有Parquet文件，可加mergeSchema解决Schema不一致问题
    df = spark.read.option("mergeSchema", "true").parquet(*paths)
    # 返回该分区的RDD数据
    return df.rdd.collect()

# 执行并行读取，得到最终的RDD
final_parquet_rdd = paths_rdd.mapPartitions(read_parquet_in_partition)

关键细节说明 #

• 分区数选择：numSlices设为文件数（100）是比较合理的，每个分区处理1个文件；如果文件大小差异大，也可以根据集群资源调整，避免小分区过多浪费资源
• 用mapPartitions而非map：map会逐个处理每个文件，频繁初始化读取器；mapPartitions批量处理一个分区内的文件，能大幅减少开销，提升性能
• 权限与依赖：确保Spark集群已配置S3访问权限（如AWS密钥、IAM角色），Executor节点能正常访问S3存储桶
• Schema兼容性：如果不同Parquet文件Schema不一致，一定要加上mergeSchema=true参数，否则读取会失败
• 替代方案（可选）：其实Spark原生支持直接并行读取多文件，spark.read.parquet(*file_paths)会自动优化并行度，如果你没有特殊需求，这个方法更简洁；但如果必须基于sc.parallelize()实现，上面的方案完全可行

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