Dagster如何在solid调用的普通Python函数中使用自定义logger

你当前的写法不生效，核心是Dagster的colored_console_logger等自定义logger是框架内部封装的实例，不会自动注册到Python标准logging模块的全局管理器中，直接通过logging.getLogger("console_logger")拿到的是未绑定任何输出规则的空logger，自然无法和context.log共用同一套日志配置。

注意：不建议通过修改Python logging全局配置的方式绑定Dagster logger，Dagster运行时为多实例隔离架构，全局绑定会导致不同运行任务的日志混淆，存在线程安全风险。

方案1：直接透传context.log参数（最推荐，适配性最强） #

不需要修改任何全局配置，只需要在调用普通函数/类时把context.log作为参数传入即可，后续你替换/新增任意Dagster自定义logger，这套逻辑都可以直接复用，不需要修改下层函数代码。

修改后的random_func.py #

def random_func(logger):
    logger.info('in random func')
    print('\nhi\n')

修改solid内调用逻辑 #

@solid
def test_logs(context):
    context.log.info("Hello, world!")
    # 直接透传上下文logger即可
    random_func(logger=context.log)

如果是普通类的场景，直接在类初始化时传入logger即可：

# 普通类示例
class CustomTool:
    def __init__(self, logger):
        self.logger = logger
    
    def run_task(self):
        self.logger.debug("类方法输出调试日志")
        self.logger.info("类方法执行完成")

# solid内调用
@solid
def test_logs(context):
    tool = CustomTool(logger=context.log)
    tool.run_task()

方案2：封装为Dagster资源（适合多solid/大量公用函数复用场景） #

如果你的项目里有大量公用函数/类都需要用到同一套logger，可以把logger封装为Dagster全局资源，避免每次都重复传参。

步骤1：定义logger资源 #

from dagster import resource

@resource
def common_logger(context):
    return context.log

步骤2：注册资源到模式定义 #

@pipeline(mode_defs=[
    ModeDefinition(
        logger_defs={"console_logger": colored_console_logger},
        # 注册logger资源
        resource_defs={"common_logger": common_logger}
    )
])
def test_pipeline():
    test_logs()

步骤3：solid中引入资源使用 #

# 通过required_resource_keys声明依赖的资源
@solid(required_resource_keys={"common_logger"})
def test_logs(context):
    context.log.info("Hello, world!")
    # 直接使用资源里的logger，也可以透传给下层函数
    random_func(logger=context.resources.common_logger)

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