如何从PySpark DataFrame提取路径并传入spark.read.parquet使用

核心结论 #

完全可以从PySpark DataFrame中提取路径用于spark.read.parquet读取，你遇到的Column is not iterable报错，是因为直接将PySpark分布式Column对象作为参数传入，Column不属于Python原生可迭代类型，无法被读取方法识别。

解决方案 #

场景1：路径总量不大（10万条路径以内） #

直接将路径列拉取到Driver端转为Python列表后传入即可，示例代码：

# 假设存储路径的DataFrame为df，路径列名为file_path
# 提取路径列、去重后转为Python本地列表
path_list = df.select("file_path").distinct().rdd.flatMap(lambda x: x).collect()

# 解包列表传入parquet读取方法
parquet_df = spark.read.parquet(*path_list)

场景2：路径总量极大（超过100万条，避免Driver内存溢出） #

先将路径输出为临时文本文件，再读取临时文件内容作为输入路径，避免全量路径占用Driver内存：

# 输出路径到临时目录，合并为1个分区避免生成多份小文件
df.select("file_path").distinct().coalesce(1).write.mode("overwrite").text("/tmp/temp_path_list")

# 读取本地临时文件得到路径列表
with open("/tmp/temp_path_list/part-00000") as f:
    path_list = [line.strip() for line in f if line.strip()]

parquet_df = spark.read.parquet(*path_list)

常见错误写法（触发报错的原因） #

# 错误：直接传入Column对象，会触发Column is not iterable报错
wrong_df = spark.read.parquet(df.select("file_path"))

注意事项 #

• 若需要标记读入数据的来源路径，可在读入后调用input_file_name()函数新增来源列：

  from pyspark.sql.functions import input_file_name
  parquet_df = parquet_df.withColumn("source_path", input_file_name())
  

• Spark原生支持路径通配符，若提取的路径包含*、?等通配符，直接传入即可正常识别
• 路径去重操作可避免重复读取同一个Parquet文件，减少不必要的资源消耗

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