Airflow 2.7.2中字符串拼接异常问题排查

问题背景 #

在Python 3.11.5搭配Airflow 2.7.2的环境中，字符串拼接逻辑本地单元测试完全正常，但在Airflow DAG执行时，日志输出会出现看似"缺失元素"的情况——实际拼接结果完整，只是连续重复的行被日志系统自动合并了。

复现代码 #

values = ['1234', '5678', 'ABC_123', 'xyz-calc', '2024-01-01',
 'NULL', '9876', 'NULL', 'example', 42, '2024-07-28T01:23:45.678',
'2024-07-28T02:34:56.789', '2024-07-28T03:45:67.890', 'user_test',
 'complete', '2024-07-28T04:56:78.901',
'2024-07-28T05:67:89.012', 'NULL',
'spark-calc-1234-driver', 'NULL', 'NULL', 'XYZ']

values_str_list = []
for value in values:
    if isinstance(value, int):
        values_str_list.append(str(value))
    elif value == 'NULL':
        values_str_list.append('NULL')
    else:
        values_str_list.append(f"'{value}'")

values_str_3 = ',\n    '.join(values_str_list)
print("Concatenated string values_str_3:")
print(values_str_3)

异常日志表现 #

日志中连续的NULL行被合并，看似缺失一个元素：

[2024-07-29, 22:12:39 UTC] {logging_mixin.py:151} INFO - Concatenated string values_str_3:
[2024-07-29, 22:12:39 UTC] {logging_mixin.py:151} INFO - 1234,
    5678,
    'ABC_123',
    'xyz-calc',
    '2024-01-01',
    NULL,
    '9876',
    NULL,
    'example',
    42,
    '2024-07-28T01:23:45.678',
    '2024-07-28T02:34:56.789',
    '2024-07-28T03:45:67.890',
    'user_test',
    'complete',
    '2024-07-28T04:56:78.901',
    '2024-07-28T05:67:89.012',
    NULL,
    'spark-calc-1234-driver',
    NULL,
    'XYZ'

但实际拼接结果完整，分割后可验证所有元素存在：

[2024-07-29, 22:12:39 UTC] {logging_mixin.py:151} INFO - Original string values_str_3: ['1234', '5678',
 "'ABC_123'", "'xyz-calc'", "'2024-01-01'",
'NULL', "'9876'", 'NULL', "'example'", '42', "'2024-07-28T01:23:45.678'",
"'2024-07-28T02:34:56.789'", "'2024-07-28T03:45:67.890'", "'user_test'",
"'complete'", "'2024-07-28T04:56:78.901'", "'2024-07-28T05:67:89.012'", 'NULL',
"'spark-calc-1234-driver'", 'NULL', 'NULL', "'XYZ'"]

原因分析 #

这是Airflow默认启用的日志行去重优化导致的。Airflow的TaskLogReader组件会自动检测并合并连续重复的日志行，减少日志冗余。当拼接字符串中出现连续相同的行（比如连续的NULL,），就会被日志系统判定为重复行并合并，造成"元素缺失"的假象。

解决方案 #

方案1：临时绕过日志去重 #

在重复元素中添加不影响业务的微小差异，让日志系统无法判定为重复行。例如，将NULL改为NULL （末尾加空格）：

elif value == 'NULL':
    values_str_list.append('NULL ')  # 加空格避免重复行合并

如果是用于SQL语句，可改为CAST('NULL' AS VARCHAR)等等效写法，既保留业务逻辑，又避免日志去重。

方案2：全局关闭日志去重 #

修改Airflow配置文件airflow.cfg，在[logging]章节下替换日志处理器，禁用去重功能：

[logging]
# 替换默认的带去重功能的TaskLogReader
task_log_reader = airflow.utils.log.file_task_handler.FileTaskHandler

修改后重启Airflow服务即可生效。

方案3：调整日志输出格式 #

避免将拼接后的字符串拆分为多行日志输出，改为单行或JSON格式打印：

# 转为单行输出
print("Concatenated string values_str_3:", values_str_3.replace('\n    ', ', '))

# 或用JSON格式输出
import json
print("Concatenated string values_str_3:", json.dumps(values_str_3))

这样日志系统不会将多行拆分为单独的日志行，也就不会触发去重逻辑。

内容的提问来源于stack exchange，提问作者Артем Суслов