如何在Python大型流水线中记录各模块及类方法的耗时？

针对你提到的Python流水线中记录类方法和模块耗时的需求，结合你正在使用的内置logging模块，我整理了几个实用的实现方案，都是低侵入、易维护的方式：

一、单个类方法的耗时记录：装饰器方案 #

装饰器是给方法加耗时记录的最优选择，不需要修改原有业务代码，直接给目标方法打上装饰器即可，还能自动识别类名和方法名。

先实现一个通用的耗时记录装饰器：

import time
import logging

# 初始化logging配置（可以放在项目入口处）
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(), logging.FileHandler('pipeline_timer.log')]
)
logger = logging.getLogger(__name__)

def log_method_time(func):
    """用于记录类方法执行耗时的装饰器"""
    def wrapper(*args, **kwargs):
        start_time = time.perf_counter()  # 高精度计时器，不受系统时间影响
        result = func(*args, **kwargs)
        elapsed_time = time.perf_counter() - start_time
        
        # 获取当前方法所属的类名
        class_name = args[0].__class__.__name__ if args else func.__qualname__.split('.')[0]
        # 输出日志
        logger.info(f"[{class_name}.{func.__name__}] 执行完成，耗时: {elapsed_time:.4f} 秒")
        return result
    return wrapper

然后给你的类方法添加这个装饰器即可：

class DataExtractor:
    @log_method_time
    def extract_from_db(self, db_config):
        # 你的业务逻辑：比如从数据库拉取数据
        time.sleep(0.6)  # 模拟耗时操作
        return "raw_db_data"

class DataCleaner:
    @log_method_time
    def clean_null_values(self, raw_data):
        # 你的业务逻辑：清洗空值
        time.sleep(0.3)
        return "cleaned_data"

每次调用这些方法时，日志里就会自动记录对应的耗时信息，比如：

2024-05-20 14:30:00,123 - main - INFO - [DataExtractor.extract_from_db] 执行完成，耗时: 0.6012 秒

二、大型流水线的模块级耗时追踪：上下文管理器方案 #

如果你的流水线是按大模块划分的（比如数据提取阶段、数据转换阶段、结果加载阶段），用上下文管理器可以更清晰地记录整个模块的总耗时，还能捕获模块执行中的异常并记录耗时。

实现一个模块级的计时器上下文管理器：

class ModuleTimer:
    """用于记录整个模块/阶段执行耗时的上下文管理器"""
    def __init__(self, module_name):
        self.module_name = module_name
        self.start_time = None

    def __enter__(self):
        self.start_time = time.perf_counter()
        logger.info(f"模块 [{self.module_name}] 开始执行")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        elapsed_time = time.perf_counter() - self.start_time
        if exc_type is None:
            logger.info(f"模块 [{self.module_name}] 执行完成，总耗时: {elapsed_time:.4f} 秒")
        else:
            logger.error(f"模块 [{self.module_name}] 执行出错，耗时: {elapsed_time:.4f} 秒，错误信息: {str(exc_val)}")
        # 返回False表示让异常继续向外抛出（根据你的需求调整）
        return False

在流水线中使用这个上下文管理器包裹整个模块的执行：

def main_pipeline():
    # 数据提取阶段
    with ModuleTimer("数据提取模块"):
        extractor = DataExtractor()
        raw_data = extractor.extract_from_db({"db": "postgres"})
    
    # 数据清洗阶段
    with ModuleTimer("数据清洗模块"):
        cleaner = DataCleaner()
        cleaned_data = cleaner.clean_null_values(raw_data)
    
    # 后续其他模块...

if __name__ == "__main__":
    main_pipeline()

这样日志里会清晰展示每个大模块的开始、结束和耗时，即使模块出错也能记录到出错前的耗时。

三、进阶优化：结构化日志（便于后续分析） #

如果需要对耗时数据做进一步分析（比如统计每个方法的平均耗时、找出瓶颈），可以把日志输出为结构化格式（比如JSON），用内置logging就能实现：

修改装饰器的日志输出逻辑：

import json

def log_method_time(func):
    def wrapper(*args, **kwargs):
        start_time = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed_time = time.perf_counter() - start_time
        
        class_name = args[0].__class__.__name__ if args else func.__qualname__.split('.')[0]
        # 构造结构化日志内容
        structured_log = {
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "component_type": "class_method",
            "class_name": class_name,
            "method_name": func.__name__,
            "elapsed_time": round(elapsed_time, 4),
            "level": "INFO"
        }
        # 转为JSON字符串输出
        logger.info(json.dumps(structured_log))
        return result
    return wrapper

这样输出的日志是标准JSON格式，后续可以用脚本或者日志分析工具快速解析统计。

四、实用注意事项 #

• 优先使用time.perf_counter()：它是Python专门用于测量短时间间隔的高精度计时器，不受系统时间调整（比如NTP同步）的影响，比time.time()更准确。
• 多线程/多进程场景：如果你的流水线是多线程或多进程的，确保logging配置了线程安全的Handler（内置的FileHandler是线程安全的，但多进程下建议使用QueueHandler+QueueListener来避免日志混乱）。
• 避免过度记录：如果某些方法会被高频调用（比如循环内的小方法），可以给装饰器加一个开关参数，控制是否记录该方法的耗时，避免日志爆炸。
• 日志分级管理：可以把耗时记录设为INFO级别，错误设为ERROR级别，这样在调试时可以通过调整日志级别快速过滤信息。

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