使用Python logging 模块优雅记录Pandas DataFrame

本教程详细阐述了如何利用Python的`logging`模块和自定义`Formatter`来高效、灵活地记录Pandas DataFrame。通过创建一个`DataFrameFormatter`，我们能够将DataFrame内容以美观、对齐的方式逐行输出到日志文件，并为每行添加标准的日志元数据（如时间戳、日志级别）。文章还将介绍如何通过`extra`参数动态控制DataFrame的打印行数和添加自定义标题，从而实现日志输出的精细化管理和高度可读性。

在数据分析和处理流程中，使用Pandas DataFrame进行数据操作是常见的实践。为了便于调试、追踪数据状态或记录中间结果，将DataFrame的内容输出到日志文件变得尤为重要。然而，直接使用logger.info(df)通常无法满足需求，它可能只输出DataFrame的内存地址或不友好的字符串表示。更常见的方法是将DataFrame转换为字符串，然后逐行打印：

import io
import logging
import pandas as pd

MAX_NUM_DF_LOG_LINES = 4

logging.basicConfig(
    datefmt = '%y%m%d %H:%M:%S',
    format = '%(asctime)s %(levelname)s: %(message)s')
logger = logging.getLogger(__name__)
logger.setLevel(level = logging.INFO)

TESTDATA="""
enzyme  regions   N   length
AaaI    all       10  238045
AaaI    all       20  170393
AaaI    captured  10  292735
AaaI    captured  20  229824
AagI    all       10  88337
AagI    all       20  19144
AagI    captured  10  34463
AagI    captured  20  19220
"""
df = pd.read_csv(io.StringIO(TESTDATA), sep='\s+')

logger.info('处理中间结果：df:')
for line in df.head(MAX_NUM_DF_LOG_LINES).to_string().splitlines():
    logger.info(line)

这种方法虽然能达到目的，但存在以下缺点：

1. 冗余性高： 每次需要记录DataFrame时，都需要重复for line in df.to_string().splitlines(): logger.log(level, line)这样的代码块。
2. 灵活性差： 难以动态控制打印的行数，或者在打印前添加自定义的描述性文字。
3. 不符合Pythonic风格： 将数据处理逻辑（DataFrame切片和字符串转换）与日志记录逻辑混杂在一起。

为了解决这些问题，Python的logging模块提供了高度可扩展的机制，其中自定义Formatter是实现优雅日志记录的关键。

立即学习“Python免费学习笔记（深入）”；

使用自定义 logging.Formatter 记录 DataFrame

logging.Formatter是负责将LogRecord对象（包含日志消息、级别、时间等信息）转换为最终字符串输出的类。通过继承并重写其format方法，我们可以定义自己的日志消息处理逻辑。

实现 DataFrameFormatter

我们将创建一个名为DataFrameFormatter的自定义格式化器，它能够识别并特殊处理pd.DataFrame类型的日志消息。

歌者PPT

歌者PPT，AI 写 PPT 永久免费

下载

import logging
import pandas as pd
import io

class DataFrameFormatter(logging.Formatter):
    """
    一个自定义的日志格式化器，用于美观地打印Pandas DataFrame。
    它能够将DataFrame内容逐行输出，并为每行添加标准的日志元数据。
    """
    def __init__(self, fmt: str, n_rows: int = 4) -> None:
        """
        初始化DataFrameFormatter。

        Args:
            fmt (str): 标准的日志格式字符串，例如 '%(asctime)s %(levelname)-8s %(message)s'。
            n_rows (int): 默认情况下打印DataFrame的行数。
        """
        self.n_rows = n_rows
        super().__init__(fmt)

    def format(self, record: logging.LogRecord) -> str:
        """
        重写format方法，处理DataFrame类型的日志消息。

        Args:
            record (logging.LogRecord): 待格式化的日志记录对象。

        Returns:
            str: 格式化后的日志字符串。
        """
        # 检查日志消息是否为DataFrame类型
        if isinstance(record.msg, pd.DataFrame):
            # 初始化一个空字符串用于构建多行日志输出
            formatted_message_lines = []

            # 动态获取打印行数：优先使用extra参数中的n_rows，否则使用实例默认值
            current_n_rows = getattr(record, 'n_rows', self.n_rows)

            # 检查是否有自定义的标题
            if hasattr(record, 'header'):
                # 暂时保存原始消息，设置消息为标题，然后格式化标题行
                original_msg = record.msg
                record.msg = record.header.strip()
                formatted_message_lines.append(super().format(record))
                record.msg = original_msg # 恢复原始消息

            # 将DataFrame的前n行转换为字符串，并按行分割
            df_string_lines = record.msg.head(current_n_rows).to_string().splitlines()

            # 逐行格式化DataFrame内容，每行都带有日志元数据
            for line in df_string_lines:
                original_msg = record.msg # 再次保存原始消息
                record.msg = line # 将当前行设置为消息
                formatted_message_lines.append(super().format(record))
                record.msg = original_msg # 恢复原始消息

            # 将所有格式化后的行连接起来，并去除末尾的换行符
            return '\n'.join(formatted_message_lines).strip()

        # 如果不是DataFrame，则使用父类的format方法进行标准处理
        return super().format(record)

DataFrameFormatter 的核心逻辑解析

1. __init__(self, fmt: str, n_rows: int = 4):

   • 接收一个标准格式字符串fmt，用于配置日志行的元数据（如时间、级别）。
   • n_rows参数设定了默认要打印的DataFrame行数。

2. format(self, record: logging.LogRecord) -> str:

   • 类型检查: if isinstance(record.msg, pd.DataFrame) 是关键，它判断当前的日志消息(record.msg)是否是一个Pandas DataFrame对象。
   • 动态行数控制: current_n_rows = getattr(record, 'n_rows', self.n_rows) 允许在每次调用logger时，通过extra参数（例如logger.info(df, extra={'n_rows': 2})）动态指定要打印的行数。如果extra中没有提供n_rows，则使用DataFrameFormatter实例初始化时设置的默认值。
   • 自定义标题: if hasattr(record, 'header') 允许通过extra={'header': "自定义标题"}在DataFrame内容之前添加一行描述性文字。
   • 逐行格式化:
     • record.msg.head(current_n_rows).to_string().splitlines() 将DataFrame的前几行转换为一个格式化好的字符串，然后按行分割。
     • 在循环中，record.msg = line 临时将LogRecord的消息内容替换为DataFrame的当前行。
     • super().format(record) 调用父类logging.Formatter的format方法，这样每一行DataFrame内容都会被加上标准的日志元数据（时间戳、日志级别等）。
     • 每次调用super().format(record)后，都会将record.msg恢复为原始的DataFrame对象，以避免对后续处理造成影响。
   • 非DataFrame消息: 如果record.msg不是DataFrame，则直接调用super().format(record)，保持其他类型日志消息的正常处理。

配置和使用示例

现在，我们来展示如何将这个自定义格式化器应用到日志系统中。

# 导入必要的库
import logging
import pandas as pd
import io

# 假设DataFrameFormatter类已经定义如上

# 创建一个DataFrame用于测试
TESTDATA="""
enzyme  regions   N   length
AaaI    all       10  238045
AaaI    all       20  170393
AaaI    captured  10  292735
AaaI    captured  20  229824
AagI    all       10  88337
AagI    all       20  19144
AagI    captured  10  34463
AagI    captured  20  19220
"""
df = pd.read_csv(io.StringIO(TESTDATA), sep='\s+')

# 1. 获取Logger实例
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG) # 设置日志级别为DEBUG，以捕获所有消息

# 2. 创建一个StreamHandler，用于将日志输出到控制台
ch = logging.StreamHandler()

# 3. 实例化自定义的DataFrameFormatter
# 定义日志格式，例如：日期时间 级别 消息
# n_rows=4 设置默认打印DataFrame的前4行
formatter = DataFrameFormatter('%(asctime)s %(levelname)-8s %(message)s', n_rows=4)

# 4. 将格式化器设置给Handler
ch.setFormatter(formatter)

# 5. 将Handler添加到Logger
logger.addHandler(ch)

# --- 示例用法 ---

logger.info('开始数据处理流程...')

# 示例1：使用默认行数和自定义标题记录DataFrame
logger.info(df, extra={'header': "这是处理后的DataFrame头部信息："})

logger.debug('这是一个普通的调试消息。')    

# 示例2：动态指定打印行数（只打印2行），不带标题
logger.info(df, extra={'n_rows': 2})

logger.warning('这是一个警告消息，数据可能存在异常。')

# 示例3：使用默认行数（4行），不带标题
logger.info(df)

logger.info('数据处理流程结束。')

预期输出示例：

2024-01-09 15:09:53,384 INFO     开始数据处理流程...
2024-01-09 15:09:53,384 INFO     这是处理后的DataFrame头部信息：
2024-01-09 15:09:53,384 INFO        enzyme   regions   N  length
2024-01-09 15:09:53,384 INFO     0   AaaI       all  10  238045
2024-01-09 15:09:53,384 INFO     1   AaaI       all  20  170393
2024-01-09 15:09:53,384 INFO     2   AaaI  captured  10  292735
2024-01-09 15:09:53,384 INFO     3   AaaI  captured  20  229824
2024-01-09 15:09:53,385 DEBUG    这是一个普通的调试消息。
2024-01-09 15:09:53,385 INFO        enzyme   regions   N  length
2024-01-09 15:09:53,385 INFO     0   AaaI       all  10  238045
2024-01-09 15:09:53,385 INFO     1   AaaI       all  20  170393
2024-01-09 15:09:53,385 WARNING  这是一个警告消息，数据可能存在异常。
2024-01-09 15:09:53,385 INFO        enzyme   regions   N  length
2024-01-09 15:09:53,385 INFO     0   AaaI       all  10  238045
2024-01-09 15:09:53,385 INFO     1   AaaI       all  20  170393
2024-01-09 15:09:53,385 INFO     2   AaaI  captured  10  292735
2024-01-09 15:09:53,385 INFO     3   AaaI  captured  20  229824
2024-01-09 15:09:53,385 INFO     数据处理流程结束。

注意事项与总结

1. 性能考虑: df.head(n).to_string()操作对于非常大的DataFrame可能会有性能开销，尤其是在频繁记录的情况下。然而，由于我们通常只记录DataFrame的头部（少量行），这种开销通常可以忽略不计。如果需要记录整个DataFrame，且DataFrame非常庞大，可能需要考虑将DataFrame保存到文件（如CSV、Parquet）并记录文件路径，而不是直接打印其内容。
2. extra 参数的灵活性: extra参数是logging模块提供的一个强大功能，它允许在LogRecord中附加任何自定义属性。我们的DataFrameFormatter正是利用了这一点，通过getattr(record, 'n_rows', self.n_rows)和hasattr(record, 'header')来访问这些动态参数。
3. 与其他Handler的兼容性: 这个自定义Formatter可以与任何logging Handler（如FileHandler、RotatingFileHandler等）配合使用，从而将格式化后的DataFrame日志输出到文件、网络或任何其他目标。
4. 日志级别控制: DataFrameFormatter完美地融入了logging模块的级别控制机制。如果logger的级别设置为INFO，那么logger.debug(df)将不会被输出，这与其他类型的日志消息行为一致。
5. Pythonic风格: 通过封装日志处理逻辑到自定义Formatter中，我们使得日志调用更加简洁和富有表达力（例如logger.info(df, extra={'header': "..."})），提升了代码的可读性和可维护性。

通过上述方法，我们不仅解决了将Pandas DataFrame优雅地输出到日志的问题，而且还通过logging模块的强大扩展性，实现了高度的灵活性和控制力，使得日志系统更加健壮和专业。