PySpark读取少量CSV耗时过长，批量读取数千CSV求解决方案

问题描述 #

初始读取性能问题 #

本地环境使用PySpark读取CSV文件，初始代码如下：

import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder \
      .master("local[1]") \
      .appName("SparkByExamples.com") \
      .getOrCreate() 

df = spark.read.csv(path+'*.csv')

仅读取两个各约1200行的CSV文件时耗时极长，需排查操作问题。

批量文件读取需求 #

尝试以下方法成功读取2个文件，但需要读取3000+个CSV文件时，寻求高效处理方案：

from pyspark.sql import SparkSession
 
spark = SparkSession.builder.appName('Read Multiple CSV Files').getOrCreate()
 
path = ['files/data1.csv',
        'files/data2.csv']
 
files = spark.read.csv(path, sep=',',
                       inferSchema=True, header=True)
 
df1 = files.toPandas()

解决方案 #

一、初始读取慢的原因及修复 #

1. 单核心限制并行能力
   代码中设置.master("local[1]")，强制Spark仅使用1个核心处理任务，完全无法发挥Spark的并行计算优势，单线程读取自然耗时久。修改为local[*]让Spark自动调用本地所有可用核心：

   spark = SparkSession.builder \
         .master("local[*]") \
         .appName("SparkByExamples.com") \
         .getOrCreate()
   

2. 未指定Schema导致额外扫描
   初始代码未设置header=True和inferSchema=True，Spark会默认将所有列识别为字符串类型，且为推断Schema会额外全量扫描文件内容，增加不必要的耗时。添加这两个参数可直接读取表头并自动推断列类型：

   df = spark.read.csv(path+'*.csv', sep=',', inferSchema=True, header=True)
   

二、3000+个CSV文件的高效处理方案 #

1. 用通配符批量读取（推荐）
   无需手动生成文件路径列表，直接用通配符匹配目标目录下所有CSV，Spark内部会自动优化文件读取的并行逻辑，避免手动维护路径的繁琐：

   # 匹配files目录下所有csv文件
   df = spark.read.csv("files/*.csv", sep=',', inferSchema=True, header=True)
   

2. 预定义Schema跳过自动推断
   文件数量极多时，inferSchema=True需要扫描所有文件推断类型，耗时会显著上升。可提前手动定义Schema，直接指定列名和数据类型，跳过推断步骤：

   from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType
   
   # 根据实际列结构自定义Schema
   custom_schema = StructType([
       StructField("id", IntegerType(), True),
       StructField("name", StringType(), True),
       StructField("value", DoubleType(), True)
   ])
   
   df = spark.read.csv("files/*.csv", sep=',', schema=custom_schema, header=True)
   

3. 避免不必要的toPandas()转换
   toPandas()会将Spark分布式DataFrame转换为本地内存中的Pandas DataFrame，当数据总量较大时，不仅耗时还可能触发内存溢出。尽量直接使用Spark API处理数据，仅在需要导出小量结果时再转换为Pandas。

4. 调整Spark配置优化性能
   针对本地环境，可适当调整内存分配等参数，提升大文件量的处理能力：

   spark = SparkSession.builder \
         .master("local[*]") \
         .appName("ReadLargeCSV") \
         .config("spark.driver.memory", "8g")  # 根据本地内存调整，如8G
         .config("spark.executor.memory", "4g")
         .getOrCreate()
   

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