PySpark：遍历partition_paths读取Parquet时如何错误处理并继续执行

你可以把原来的列表推导式改成带异常捕获的循环，逐个尝试读取路径，跳过无效的，只保留能成功读取的DataFrame：

from functools import reduce
from pyspark.sql import DataFrame
from pyspark.sql.utils import AnalysisException  # Spark路径不存在等问题会抛出这个异常

dfs = []
for path in partition_paths:
    try:
        df = spark.read.parquet(path)
        dfs.append(df)
    except AnalysisException as e:
        # 这里可以根据需要记录日志，或者打印错误提示
        print(f"读取路径 {path} 失败: {str(e)}")
        continue
    except Exception as e:
        # 捕获其他可能的异常，比如权限问题、文件损坏等
        print(f"处理路径 {path} 时发生未知错误: {str(e)}")
        continue

# 合并前先判断是否有有效DataFrame，避免空列表调用reduce报错
if dfs:
    df = reduce(DataFrame.unionAll, dfs)
else:
    # 没有有效数据时，可以返回空DataFrame或者根据业务逻辑处理
    df = spark.createDataFrame([], schema=None)  # 或者指定你的默认schema

说明： #

• 用try-except块包裹每个路径的读取操作，专门捕获Spark的AnalysisException（这是路径不存在、格式错误等常见问题抛出的异常），同时保留通用异常捕获处理其他意外情况
• 读取失败时打印错误信息（也可以换成日志工具记录，比如logging.error），然后跳过当前路径，继续处理下一个
• 合并前检查dfs是否为空，避免reduce处理空列表抛出TypeError

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