Spark高效批量加载多txt文件并生成合并DataFrame方案问询

针对你需要快速处理大量独立小型.txt文件，生成包含file_name、start_time、end_time、duration的单一DataFrame的需求，你最终采用的方案已经是非常高效的分布式处理方式了，完全避开了逐行循环的低效问题，下面我来详细拆解下这个方案的关键细节：

核心实现代码 #

from pyspark.sql.functions import split, reverse, input_file_name, col
from pyspark.sql.types import StructField, FloatType, StructType

original_schema = [StructField("Start", FloatType(), True), StructField("End", FloatType(), True)]
data_structure = StructType(original_schema)

df = self.spark_session.read.\
    csv(path=PATH_FILES+'\\*.txt', header=False, schema=data_structure, sep='\t').\
    withColumn("Filename", reverse(split(input_file_name(), "/")).getItem(0) ).\
    withColumn("duration", col("End") - col("Start"))

df.show(20, False)

关键优化点解析 #

• 提前指定Schema：手动定义StructType并传给read.csv，避免Spark自动推断数据类型的额外开销。对于大量小文件来说，类型推断需要扫描每个文件的部分数据，提前指定Schema能大幅提升加载速度。
• 分布式获取文件名：使用Spark内置的input_file_name()函数，能在分布式环境下准确获取每行数据所属的文件完整路径，再通过split+reverse提取纯文件名，这个操作是在Spark集群上并行执行的，比单节点遍历文件系统高效得多。
• 向量式计算时长：直接用col("End") - col("Start")生成duration列，Spark会把这个运算转化为分布式的向量计算，完全避开了逐行处理的单线程瓶颈。

为什么不用textFile()或wholeTextFiles()？ #

这两个RDD API虽然也能加载文件，但需要手动处理每行的分割、类型转换，还要自己关联文件名，代码复杂度更高，而且性能不如DataFrame API的内置优化。你用的spark.read.csv本质上是把每个.txt文件当成无表头的CSV文件处理，完美适配你每行两列的结构化数据，是最适合的选择。

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