如何在虚拟环境中运行Airflow PythonOperator

Question

我目前正在使用 BashOperator 执行几个 python 文件。这让我可以灵活地轻松选择Python虚拟环境。

from airflow import DAG
from airflow.operators.bash_operator import BashOperator

default_args = {
   'owner': 'airflow',
    'depends_on_past': False,
    ...}

dag = DAG('python_tasks', default_args=default_args, schedule_interval="23 4 * * *")

t1 = BashOperator(
                 task_id='task1',
                bash_command='~/anaconda3/envs/myenv/bin/python 
                              /python_files/python_task1.py',
                 dag=dag)

如何使用 PythonOperator 实现类似的效果？

from airflow.operators.bash_operator import PythonOperator
import python_files.python_task1

python_task = PythonOperator(
              task_id='python_task',
              python_callable=python_task1.main,
             dag=dag)

我假设PythonOperator将使用系统python环境。 我发现 Airflow 有 PythonVirtualenvOperator，但这似乎是通过使用指定的要求动态创建一个新的虚拟环境来工作的。我更愿意使用已正确配置的现有版本。 如何使用指定的Python路径运行PythonOperator？

Answer 1

我的解决方法是使用 Bash Operator 来调用

/path/to/project/venv/bin/python my.py

Answer 2

首先要做的事情是：您（通常）不应该依赖您的操作员预先存在的资源。您的操作员应该是可移植的，因此使用长期存在的 virtualenvs 在某种程度上违反了这一原则。话虽如此，这并不是什么大问题，就像您必须将软件包预安装到全局环境一样，您可以预烘焙一些环境。或者，您可以让 Operator 创建环境，然后后续的 Operator 可以重用它 - 我认为这是最简单也是最危险的方法。

实现“virtualenv 缓存”应该不难。阅读

PythonVirtualenvOperator

的执行方法的实现：

def execute_callable(self):
    with TemporaryDirectory(prefix='venv') as tmp_dir:
        ...
        self._execute_in_subprocess(
            self._generate_python_cmd(tmp_dir,
                                      script_filename,
                                      input_filename,
                                      output_filename,
                                      string_args_filename))
        return self._read_result(output_filename)

所以看起来它没有显式删除 virtualenv （它依赖于

TemporaryDirectory

来做到这一点）。您可以子类

PythonVirtualenvOperator

并简单地使用您自己的上下文管理器来重用临时目录：

import glob

@contextmanager
def ReusableTemporaryDirectory(prefix):
    try:
        existing = glob.glob('/tmp/' + prefix + '*')
        if len(existing):
            name = existing[0]
        else:
            name = mkdtemp(prefix=prefix)
        yield name
    finally:
        # simply don't delete the tmp dir
        pass

def execute_callable(self):
    with ReusableTemporaryDirectory(prefix='cached-venv') as tmp_dir:
        ...

当然，您可以去掉

try-finally

中的

ReusableTemporaryDirectory

并放回通常的

suffix

和

dir

参数，我做了最小的更改，以便于与原始

TemporaryDirectory

类进行比较。

这样，您的 virtualenv 将不会被丢弃，但 Operator 最终将安装更新的依赖项。

Answer 3

使用PythonVirtualenvOperator。您需要提供要运行的python函数和虚拟环境的requirements.txt。