Pandas重复值检测代码解析及异常抛出实现问询

先拿这类代码的常见实现来拆解：

import pandas as pd
from io import StringIO
import logging

# 初始化日志记录器
string_io_logger = logging.getLogger("duplicate_check")
string_io_logger.setLevel(logging.ERROR)
handler = logging.StreamHandler(StringIO())
string_io_logger.addHandler(handler)

# 示例数据
df = pd.DataFrame({"映射字段": ["A", "B", "A", "C"]})

# 检测重复值
duplicate_mask = df["映射字段"].duplicated(keep=False)
duplicate_rows = df[duplicate_mask]

if not duplicate_rows.empty:
    error_msg = f"检测到映射字段存在重复值：{duplicate_rows['映射字段'].tolist()}"
    string_io_logger.error(error_msg)
    # 原注释说要抛错误，但这里没写对应代码

它的执行步骤是：

1. 标记重复行：用df["映射字段"].duplicated(keep=False)生成一个布尔序列，所有重复出现的行都会被标记为True（如果用默认的keep='first'，只会标记除第一次出现外的重复行）。
2. 筛选重复数据：用这个布尔序列过滤原DataFrame，得到所有包含重复值的行duplicate_rows。
3. 判断并记录错误：检查duplicate_rows是否为空，若不为空就把错误信息写入日志，但因为没有抛出异常的代码，程序会继续执行后续逻辑，不会中断。

只需要在记录日志的代码后面，添加raise语句抛出异常即可，示例如下：

if not duplicate_rows.empty:
    error_msg = f"检测到映射字段存在重复值：{duplicate_rows['映射字段'].tolist()}"
    string_io_logger.error(error_msg)
    # 抛出ValueError异常中断流程
    raise ValueError(error_msg)

如果需要更贴合业务场景的异常类型，也可以自定义异常类：

# 自定义异常类
class DuplicateMappingError(Exception):
    """映射字段重复时抛出的自定义异常"""
    pass

# 在检测到重复时抛出
if not duplicate_rows.empty:
    error_msg = f"检测到映射字段存在重复值：{duplicate_rows['映射字段'].tolist()}"
    string_io_logger.error(error_msg)
    raise DuplicateMappingError(error_msg)

抛出异常后，程序会立刻终止当前执行流程，后续代码不会再运行；如果需要捕获这个异常做额外处理，可以在调用这段检测逻辑的地方用try-except包裹。

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