Airflow中xcom_push函数的key与value参数定义问题求助

嘿，我来帮你理清楚Airflow中XCom的正确用法，你的代码里确实在xcom_push的参数设置上踩了几个小坑，咱们一步步来修正：

先说说你当前代码的问题 #

1. key参数类型错误：xcom_push的key需要是字符串类型的标识，用来让后续任务精准定位要获取的XCom数据。你现在传的是db_con（数据库连接对象），这完全不符合要求。
2. value参数选得不对：你把函数db_log本身作为value，这显然不是你想要传递的内容；而且就算你想传db_con，数据库连接对象是不可序列化的——XCom底层是把数据存在Airflow的元数据库里，必须是能被pickle序列化的类型（比如字符串、字典、列表这些），直接传连接对象会报错。

正确的用法示例 #

首先要明确：XCom适合传递小量、可序列化的数据，比如数据库连接参数、查询结果等，绝对不能传递数据库连接这类“活”的对象。下面给你两种常见场景的修正代码：

场景1：传递数据库连接参数（供后续任务重新建立连接）

import psycopg2

def db_log(**context):
    # 把数据库连接参数封装成可序列化的字典
    db_connection_params = {
        "dbname": "name",
        "user": "user",
        "password": "pass",
        "host": "host",
        "port": "5439",
        "sslmode": "require"
    }
    
    # 当前任务需要使用连接的话，在这里建立并使用
    db_con = psycopg2.connect(**db_connection_params)
    # （这里可以加你的数据库操作逻辑，比如执行SQL、写入日志等）
    
    # 获取task_instance对象
    task_instance = context['task_instance']
    
    # 正确调用xcom_push：key是字符串标识，value是可序列化的字典
    task_instance.xcom_push(key='db_conn_params', value=db_connection_params)
    
    # 记得关闭连接，避免资源泄漏
    db_con.close()
    
    # 额外提一句：Airflow会自动把函数return的值以key='return_value'存入XCom
    # 所以如果只需要传递这一个参数，也可以不用显式调用xcom_push，直接return即可
    return db_connection_params

场景2：传递数据库查询结果

如果你是想把当前任务的数据库查询结果传给后续任务，可以这样写：

import psycopg2

def db_log(**context):
    db_params = {
        "dbname": "name",
        "user": "user",
        "password": "pass",
        "host": "host",
        "port": "5439",
        "sslmode": "require"
    }
    
    db_con = psycopg2.connect(**db_params)
    cur = db_con.cursor()
    
    # 执行你的查询逻辑
    cur.execute("SELECT id, log_content FROM operation_logs LIMIT 100")
    query_results = cur.fetchall()  # 这是一个可序列化的列表
    
    # 存入XCom
    task_instance = context['task_instance']
    task_instance.xcom_push(key='recent_operation_logs', value=query_results)
    
    # 清理资源
    cur.close()
    db_con.close()
    
    return query_results

后续任务如何获取XCom数据 #

比如在另一个PythonOperator任务里，你可以这样获取上面存入的参数：

def use_db_params(**context):
    task_instance = context['task_instance']
    # 通过key和任务ID获取XCom数据
    db_params = task_instance.xcom_pull(key='db_conn_params', task_ids='你的db_log任务ID')
    # 重新建立连接
    db_con = psycopg2.connect(**db_params)
    # 后续操作...

重要注意事项 #

• 不要尝试传递数据库连接、文件句柄这类“活”的对象，它们无法被序列化，会触发报错。
• 如果要传递大量数据（比如上万条查询结果），别用XCom——应该把数据存在外部存储（比如S3、专门的结果表），然后用XCom传递存储路径或表名这类标识。

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