PySpark读取多文件：兼容CSV与TSV多种分隔符的方法

核心思路 #

PySpark原生csv读取器默认仅支持单一分隔符，但可以通过正则表达式匹配多分隔符或先固定Schema再读取的方式，同时兼容逗号（CSV）和制表符（TSV）分隔的文件。需要注意的是，自动推断Schema时若采样到制表符文件，可能出现列推断错误，因此更稳妥的方式是先明确Schema再读取。

方法1：使用正则表达式作为分隔符 #

PySpark的sep参数支持正则表达式，直接指定匹配逗号或制表符即可：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("MultiDelimiterLoader").getOrCreate()

# 用正则匹配逗号或制表符作为分隔符
df = spark.read.option("sep", r",|\t") \
               .option("header", "true") \
               .option("inferSchema", "true") \
               .csv("/path/to/your/target/folder")

df.show()

提示：如果开启inferSchema，Spark会随机采样文件推断Schema。若采样到制表符文件，可能导致Schema推断偏差，建议先从已知正常的CSV文件提前获取Schema。

方法2：预定义Schema后读取（推荐） #

1. 先从一个标准CSV文件推断并保存Schema：

# 读取单个正常CSV文件获取Schema
sample_df = spark.read.option("header", "true") \
                      .option("inferSchema", "true") \
                      .csv("/path/to/sample.csv")
fixed_schema = sample_df.schema

2. 复用该Schema读取整个文件夹，同时指定多分隔符：

df = spark.read.option("sep", r",|\t") \
               .option("header", "true") \
               .schema(fixed_schema) \
               .csv("/path/to/your/target/folder")

这种方式完全规避了Schema推断错误的问题，稳定性和效率都更高。

方法3：自定义文件读取逻辑（小数据集适用） #

如果需要更精细的控制，可以遍历文件夹内的每个文件，判断分隔符后再读取：

import os
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("CustomMultiDelimiter").getOrCreate()

def read_single_file(file_path, target_schema):
    # 读取文件首行判断分隔符
    with open(file_path, 'r') as f:
        first_line = f.readline()
        sep = ',' if first_line.count(',') > first_line.count('\t') else '\t'
    
    return spark.read.option("sep", sep) \
                     .option("header", "true") \
                     .schema(target_schema) \
                     .csv(file_path)

# 获取文件夹下所有文件路径
folder_path = "/path/to/your/target/folder"
file_list = [os.path.join(folder_path, f) for f in os.listdir(folder_path) if os.path.isfile(os.path.join(folder_path, f))]

# 逐个读取并合并数据
final_df = spark.createDataFrame([], fixed_schema)
for file in file_list:
    temp_df = read_single_file(file, fixed_schema)
    final_df = final_df.union(temp_df)

final_df.show()

该方法适合数据量较小的场景，缺点是需要遍历本地文件系统，效率略低。

注意事项 #

• 确保所有文件的列数、列名完全一致，否则合并时会抛出Schema不匹配的错误。
• 若文件无表头，需移除option("header", "true")，并调整Schema的字段名与文件列对应。
• 正则分隔符的性能略低于单一分隔符，大数据集优先选择方法2。

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