如何让Pandas在部分行操作异常时告警并继续处理有效行？

针对少量异常数据导致pd.apply崩溃的问题，以下是简洁通用的实现方案，既能让脚本继续处理正常数据，又能捕获并报告错误行：

核心思路 #

通过带异常捕获的处理函数封装业务逻辑，执行时分离成功/失败数据，同时记录错误信息；无需提前预判所有异常类型，覆盖绝大多数意外情况。

1. 通用异常捕获装饰器 #

先写一个装饰器，把异常捕获逻辑和业务逻辑解耦，避免重复代码：

import pandas as pd
import warnings

def catch_process_exceptions(func):
    def wrapper(row):
        try:
            # 执行原业务逻辑
            return func(row), None
        except Exception as e:
            # 生成错误信息并发出警告
            error_msg = f"行索引 {row.name} 处理失败: {str(e)}"
            warnings.warn(error_msg)
            return None, error_msg
    return wrapper

2. 封装业务处理函数 #

用装饰器标记你的数据处理逻辑：

@catch_process_exceptions
def process_single_row(row):
    # 这里替换成你的实际处理代码，比如：
    return row['value1'] * row['value2'] + row['value3']

3. 批量执行并分离数据 #

将函数应用到DataFrame，拆分结果和错误信息，过滤出正常数据继续处理：

# 示例输入数据（包含异常行）
df = pd.DataFrame({
    'value1': [10, 20, 'invalid', 40],
    'value2': [5, 6, 7, 8],
    'value3': [1, 2, 3, 4]
})

# 应用处理函数，生成结果列和错误列
df['processed'], df['error'] = zip(*df.apply(process_single_row, axis=1))

# 分离成功/失败数据集
success_df = df[df['error'].isna()].drop(columns=['error'])
failed_df = df[df['error'].notna()][['error']]

# 输出错误报告
if not failed_df.empty:
    print(f"\n⚠️ 共 {len(failed_df)} 行处理失败：")
    for msg in failed_df['error']:
        print(f"- {msg}")

# 后续用success_df执行数据库更新等操作
# 比如：success_df.to_sql(...)

方案优势 #

• 无侵入性：业务逻辑无需修改，仅通过装饰器添加异常处理
• 全量覆盖：捕获所有未预见的异常，避免脚本崩溃
• 可追溯性：保留错误行的索引和详情，方便排查问题
• 不中断流程：正常数据继续处理，不会因少量异常导致全量任务失败

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