如何通过编程方式获取Airflow日志并存储为字符串变量？

你之前的代码里ti.log是Airflow用于写入日志的Logger对象，不是已生成的日志内容，所以打印出来是Logger实例的描述。要获取任务实例的日志内容，需要用Airflow的日志阅读器来读取，具体实现如下：

import logging
from airflow.models import TaskInstance

def fetch_logs(context):
    ti: TaskInstance = context['ti']
    # 获取当前任务实例的日志阅读器
    log_reader = ti.log_reader
    # 读取对应重试次数的日志，这里使用当前任务的重试次数
    logs_content = log_reader.read_log(
        task_instance=ti,
        try_number=ti.try_number
    )
    # logs_content即为日志的字符串形式，可直接用于后续逻辑
    logging.info("获取到的任务日志:\n%s", logs_content)
    return logs_content

补充说明： #

• read_log方法默认读取对应任务实例指定重试次数的全部日志，你也可以通过添加start_line和end_line参数来读取指定范围的日志内容
• 该方法适配Airflow 2.x版本，不同版本的日志阅读器API可能略有差异，若使用旧版本可查看对应版本的官方文档调整参数

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