Python Flink本地执行日志配置问题：无法设级别及算子内日志不显示

问题分析 #

本地直接执行Python Flink脚本时，会遇到两个核心日志问题：

• Python标准logging模块设置的日志级别不生效，INFO级日志被屏蔽
• 算子内的日志（包括WARNING级）无法输出

这是因为：

• 顶层Python代码的日志在主进程输出，而算子代码运行在Flink TaskManager启动的Python子进程中，两者日志环境隔离
• Python日志会被Flink的Java日志框架（Log4j/Logback）接管，单纯配置Python侧的日志级别无法穿透到Java侧的过滤规则

解决方案 #

1. 配置Flink的Log4j控制台日志规则 #

找到Flink安装目录下的conf/log4j-console.properties，修改以下配置项：

# 根日志级别设为DEBUG，确保所有级别日志能进入控制台
rootLogger.level = DEBUG
# 控制台Appender的阈值设为DEBUG，不拦截低级别日志
appender.console.filter.threshold.level = DEBUG
# 专门针对Flink Python模块的日志配置
logger.pyflink.name = org.apache.flink.python
logger.pyflink.level = DEBUG
logger.pyflink.additivity = false
logger.pyflink.appenderRef.console.ref = Console

如果不想修改全局配置，可在启动脚本时通过JVM参数指定自定义配置文件：

python test.py -Dlog4j.configuration=file:///path/to/your/custom-log4j.properties

2. 在Python代码中适配Flink日志桥接 #

修改代码，确保Python日志能正确转发到Flink的Java日志框架：

import logging
from pyflink.common import Types, Configuration
from pyflink.datastream import StreamExecutionEnvironment

# 配置Python基础日志级别
logging.basicConfig(level=logging.DEBUG)
# 使用Flink Python专属日志器，确保日志能被桥接
logger = logging.getLogger("org.apache.flink.python")
logger.setLevel(logging.DEBUG)

print("PRINT is visible")
logger.info("INFO should be visible now")
logger.warning("WARNING is visible")

def test_operator(n: int) -> int:
    print("PRINT is visible")
    logger.info("INFO inside operator is visible now")
    logger.warning("WARNING inside operator is visible now")
    return n

# 配置执行环境，开启Python日志转发
config = Configuration()
config.set_string("python.log.level", "DEBUG")
env = StreamExecutionEnvironment.get_execution_environment(config)

source = env.from_collection(range(5), type_info=Types.INT())
source.map(test_operator).print()
env.execute()

3. 验证效果 #

重新运行脚本后，应该能看到：

• 所有print语句的输出
• 顶层和算子内的INFO、WARNING级日志

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