加载Parquet到Delta表时自动提取顶层目录名作为列 免循环遍历万级文件夹

核心逻辑 #

完全不需要逐目录/逐文件做for循环读取，直接利用Spark内置的文件元数据能力，单次批量匹配所有目标文件，在分布式计算阶段自动溯源提取目录名作为Source列，性能远高于循环逐读的方式。

具体实现步骤 #

1. 一次性批量匹配所有目标parquet文件
   直接用通配符规则覆盖所有需要读取的路径，不需要遍历上万个SRC目录：

   from pyspark.sql.functions import input_file_name, regexp_extract
   # 通配匹配所有SRC目录下的parquet文件，触发单次分布式目录扫描
   raw_df = spark.read.parquet("/SRC*/trialbucket=*/*.parquet")
   

2. 自动提取Source溯源列
   用Spark内置的input_file_name()函数，可以在数据读取阶段直接拿到每一行数据对应的源文件全路径，之后通过正则提取需要的目录段作为Source列，全程不需要循环：

   final_df = raw_df.withColumn(
       "Source",
       # 按照实际路径结构调整匹配规则，这里示例匹配路径中/SRCxx格式的目录段
       regexp_extract(input_file_name(), r"/(SRC\d+)/", 1)
   )
   

   如果需要提取的Source目录没有固定的SRC开头命名规则，只需要调整正则匹配规则，即可适配任意路径段的提取需求，不需要改变整体的批量读取逻辑。提取完成后直接写入Delta表即可，整个流程的列计算是在Executor端并行完成的，没有额外性能开销。

针对分区目录的优化 #

你给出的路径结构是标准的Hive/Delta分区格式（trialbucket=数字），Spark支持自动识别这类分区列：

• 读取时通过basePath参数指定根路径，Spark会自动探测全部分区值，不需要手动给分区列赋值
• 遇到上万个目录的大规模文件列表扫描场景，可以开启并行分区探测配置，进一步降低扫描耗时：

  spark.conf.set("spark.sql.sources.parallelPartitionDiscovery.parallelism", "1024")
  

原循环方案的性能损耗原因 #

逐目录循环读取的实现方式，每一次读取都会触发一次独立的Spark任务调度，上万个目录就会产生上万个Job的调度开销，同时没有文件元信息的批量复用。换成单次批量读取的模式后，整个流程仅需一次Job调度，文件扫描、数据解析、列裁剪全流程分布式并行执行，整体耗时通常能降低1~2个数量级。

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