本文介绍在 pyspark 中使用 try-except 机制批量读取动态 parquet 路径列表时的异常处理方法，确保单个路径读取失败不影响整体流程，并最终安全合并所有成功加载的数据框。
本文介绍在 pyspark 中使用 try-except 机制批量读取动态 parquet 路径列表时的异常处理方法，确保单个路径读取失败不影响整体流程，并最终安全合并所有成功加载的数据框。
在实际数据湖（Data Lake）场景中，我们常需根据动态生成的路径列表（如按日期、分区字段等）批量读取 Parquet 文件。若直接使用列表推导式（如 [spark.read.parquet(p) for p in partition_paths]），任一路径不存在或格式错误将导致整个任务中断——这显然不符合生产环境对容错性和鲁棒性的要求。
正确的做法是显式遍历路径列表，对每个读取操作单独捕获异常，仅保留成功加载的 DataFrame：
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"✗ Failed to load {path}: {type(e).__name__} - {str(e)}")
# 可选：记录日志、上报监控或写入失败路径清单
continue
# 合并非空结果集（注意：若 dfs 为空，unionAll 会报错）
if not dfs:
raise ValueError("No valid DataFrame loaded from any partition path.")
# 使用 unionByName 更健壮（推荐替代 unionAll，后者已弃用）
df = reduce(lambda a, b: a.unionByName(b, allowMissingColumns=True), dfs)
⚠️ 关键注意事项：
- 避免使用 unionAll：该方法自 Spark 3.0 起已被标记为弃用（deprecated），应改用 unionByName(..., allowMissingColumns=True)，它能自动对齐列名并容忍缺失列，大幅提升兼容性；
- 空列表防护：务必检查 dfs 是否为空，否则 reduce 将抛出 TypeError；
- 异常细化：生产环境中建议捕获具体异常（如 AnalysisException、IllegalArgumentException），而非宽泛的 except:，便于精准诊断问题根源；
- 性能优化：若路径数量极大，可考虑引入并发控制（如线程池限流）或异步读取（需配合 Spark 的 spark.sql.adaptive.enabled 等配置）；
- 可观测性增强：在 except 块中添加日志级别（如 logging.warning）、失败路径持久化或指标上报，有助于运维排查。
通过上述结构化异常处理，你既能保障批处理流程的连续性，又能获得清晰的执行反馈，真正实现“失败隔离、成功聚合”的稳健数据集成模式。
