如何在Prefect 2流中将stdout和stderr转发至日志记录器？

首先明确：Prefect v2确实没有原生支持将stdout/stderr直接转发到内置日志记录器的功能，这个特性在从v1升级时被移除了。

要实现你需要的「保留原有print输出到stdout/stderr，同时把stdout消息转成logger.info、stderr转成logger.warning」的需求，可以通过自定义标准流重定向的方式实现，具体做法如下：

• 自定义一个流处理类，同时将内容写入原始的stdout/stderr和Prefect日志记录器
• 在流或任务初始化时，替换sys.stdout和sys.stderr为自定义类实例

示例代码：

import sys
import logging
from prefect import flow, get_run_logger

class StreamToLogger:
    def __init__(self, logger, log_level=logging.INFO):
        self.logger = logger
        self.log_level = log_level
        self.original_stream = sys.stdout if log_level == logging.INFO else sys.stderr

    def write(self, buf):
        # 先把内容输出到原始流，保证向后兼容
        self.original_stream.write(buf)
        self.original_stream.flush()
        # 逐行写入Prefect日志
        for line in buf.strip().splitlines():
            self.logger.log(self.log_level, line.strip())

    def flush(self):
        self.original_stream.flush()

@flow
def legacy_etl_flow():
    logger = get_run_logger()
    # 替换标准流
    sys.stdout = StreamToLogger(logger, logging.INFO)
    sys.stderr = StreamToLogger(logger, logging.WARNING)

    # 原有ETL代码里的print语句
    print("ETL数据提取开始")
    print("数据格式校验失败", file=sys.stderr)

    # 可选：任务结束后恢复原始流，避免影响后续代码
    sys.stdout = sys.stdout.original_stream
    sys.stderr = sys.stderr.original_stream

if __name__ == "__main__":
    legacy_etl_flow()

这个方案的好处是完全不需要修改原有ETL作业里的print语句，既能保持原有输出行为，又能让消息同步到Prefect Orion UI的日志系统中，分别以info和warning级别展示。如果需要在单个任务级别做适配，也可以把流替换的逻辑放到任务函数内部，实现更细粒度的控制。

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