本文介绍如何在 pyspark 中使用 try-except 机制遍历动态 parquet 路径列表，优雅跳过不存在或不可读的路径，避免程序中断，确保后续 dataframe 合并操作正常执行。
本文介绍如何在 pyspark 中使用 try-except 机制遍历动态 parquet 路径列表，优雅跳过不存在或不可读的路径，避免程序中断，确保后续 dataframe 合并操作正常执行。
在实际数据湖（Data Lake）场景中，常需根据运行时生成的路径列表（如按日期、分区字段构造的 partition_paths）批量读取 Parquet 文件。若直接使用列表推导式（如 [spark.read.parquet(p) for p in partition_paths]），任一路径失效（如路径不存在、权限不足、Schema 不兼容等）将立即抛出异常（如 AnalysisException 或 IllegalArgumentException），导致整个流程中断，无法继续加载其余有效路径。
为实现容错式批量读取，推荐改用显式 for 循环配合异常捕获：
from functools import reduce
from pyspark.sql import DataFrame
dfs = []
for path in partition_paths:
try:
df = spark.read.parquet(path)
dfs.append(df)
print(f"✅ Successfully loaded: {path}")
except Exception as e:
print(f"⚠️ Skipped invalid path '{path}': {type(e).__name__} - {str(e)[:100]}")
# 合并所有成功加载的 DataFrame
if dfs:
# 注意：unionAll 已在 Spark 3.0+ 中弃用，推荐使用 union（自动去重）或 unionByName（按列名对齐）
result_df = reduce(DataFrame.union, dfs)
# 或更健壮的方式（要求 Schema 一致）：
# result_df = dfs[0].union(*dfs[1:])
else:
raise ValueError("No valid Parquet paths were loaded — check partition_paths and storage permissions.")
? 关键注意事项：
- 不要使用裸 except: —— 应捕获具体异常（如 pyspark.sql.utils.AnalysisException）或至少保留 Exception，避免掩盖编程错误；
- 检查 dfs 非空再合并 —— 防止 reduce 在空列表上触发 TypeError；
- 优先使用 union 而非 unionAll —— 后者自 Spark 3.0 起已标记为 deprecated，union 行为与其一致（不校验列名），而 unionByName() 更安全（按列名而非位置合并）；
- 日志增强可观测性 —— 记录跳过的路径及具体错误，便于运维排查；
- 若需进一步提升鲁棒性，可添加重试逻辑、路径预检（spark.sparkContext._jvm.org.apache.hadoop.fs.FileSystem.get(...).exists(...)）或异步并发读取（配合线程池/ThreadPoolExecutor，注意 Spark Context 线程安全性）。
通过该模式，你可在数据源不稳定或路径配置偶发错误时，保障主流程持续运行，真正实现“失败隔离、尽力而为”的生产级数据加载策略。
