如何将XCom值作为参数传递给Airflow的PythonVirtualenvOperator？

在Airflow 2.2.4版本中，PythonVirtualenvOperator的requirements参数未启用Jinja模板渲染，直接传入{{ ti.pull('read_requirements') }}这类模板字符串时，会被当作纯文本处理，无法解析上游任务返回的依赖列表。

方案1：手动创建虚拟环境替代PythonVirtualenvOperator #

放弃使用官方的PythonVirtualenvOperator，改用普通PythonOperator，在任务代码中手动调用virtualenv模块创建环境、安装依赖并执行目标逻辑，这样可以直接通过XCom获取上游的依赖列表，完全控制流程。

代码示例：

from airflow import DAG
from airflow.decorators import task
from airflow.operators.python import PythonOperator
import virtualenv
import os
import tempfile

with DAG("crawler", schedule_interval=None) as dag:

    @task(task_id="read_requirements")
    def read_requirements():
        requirements_file_path = "requirements.txt"
        # 读取并清理依赖列表（去除换行符、空行）
        with open(requirements_file_path, "r") as f:
            requirements = [line.strip() for line in f if line.strip()]
        return requirements

    def run_crawler_with_virtualenv(**context):
        # 从XCom拉取上游返回的依赖列表
        requirements = context["ti"].xcom_pull(task_ids="read_requirements")
        
        # 创建临时目录作为虚拟环境路径
        with tempfile.TemporaryDirectory() as venv_dir:
            # 初始化虚拟环境
            virtualenv.create_environment(venv_dir)
            # 安装依赖
            pip_install_cmd = f"{venv_dir}/bin/pip install {' '.join(requirements)}"
            os.system(pip_install_cmd)
            
            # 定义要在虚拟环境中执行的代码
            crawler_code = """
print("hello_world")
"""
            # 激活虚拟环境并执行代码
            venv = virtualenv.VirtualEnv(venv_dir)
            venv.run(crawler_code)

    crawler_task = PythonOperator(
        task_id="crawler",
        python_callable=run_crawler_with_virtualenv,
        provide_context=True
    )

    read_requirements() >> crawler_task

方案2：自定义支持模板的PythonVirtualenvOperator子类 #

通过继承官方PythonVirtualenvOperator，将requirements字段加入模板支持列表，使其能解析Jinja表达式。

代码示例：

from airflow import DAG
from airflow.decorators import task
from airflow.operators.python import PythonVirtualenvOperator

class TemplatedPythonVirtualenvOperator(PythonVirtualenvOperator):
    # 将requirements添加到模板字段，启用Jinja渲染
    template_fields = (*PythonVirtualenvOperator.template_fields, "requirements")

with DAG("crawler", schedule_interval=None) as dag:

    @task(task_id="read_requirements")
    def read_requirements():
        requirements_file_path = "requirements.txt"
        with open(requirements_file_path, "r") as f:
            requirements = [line.strip() for line in f if line.strip()]
        # 将列表转为空格分隔的字符串，适配requirements参数格式
        return " ".join(requirements)

    def crawler():
        print("hello_world")

    crawler_task = TemplatedPythonVirtualenvOperator(
        task_id="crawler",
        python_callable=crawler,
        requirements="{{ ti.xcom_pull(task_ids='read_requirements') }}",
    )

    read_requirements() >> crawler_task

注意事项 #

• 方案2中，上游任务需将依赖列表转为空格分隔的字符串，因为PythonVirtualenvOperator的requirements参数支持字符串（空格分隔）或列表格式，模板渲染后会得到字符串，直接传入即可。
• 确保Airflow工作目录能访问到requirements.txt，建议使用绝对路径避免路径问题。

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