MWAA - 气流 - PythonVirtualenvOperator 需要 virtualenv

Question

我正在使用AWS的 mwaa服务（2.2（2.2）。 2）运行各种DAG，其中大多数是使用标准的Pythononerator类型实现的。我将DAG与任何共享要求一起将DAG捆绑到S3存储桶中，然后将MWAA指向相关对象＆amp;版本。到目前为止，一切运行顺利。

我现在想使用 pythonvirtualenvoperator 类型，这是一个奇怪的类型，开箱即用。我正在关注他们的指南使用自定义插件，但继续从气流中收到错误，在仪表板的顶部显示了大红色写作：

dag导入错误（1） ... ... ... AirFlowException：Pythonvirtualenvoperator需要Virtualenv，请安装它。

我已经确认该插件确实是通过气流拾取的（我看到它在管理员屏幕中引用了它），并且为了避免疑问，我正在使用AWS在其示例中提供的DAG中提供的确切代码。 AWS对此的文档很轻，我还没有偶然发现任何社区讨论。

从AWS的文档中，我们希望该插件在处理任何DAG之前都可以在启动过程中运行。该插件本身似乎有效地重写了VENV命令以使用PIP安装的版本，而不是安装在机器上的版本，但是我很难以我期望的顺序验证事情正在发生。任何关于调试实例行为的指示都将不胜感激。

有人面临类似的问题吗？ MWAA文档中是否有需要解决的差距？我是否想念一些非常明显的东西？

可能是相关的，但是我确实在调度程序的日志中看到了此警告，这可能表明MWAA为什么努力解决依赖性？

警告：脚本VirtualEnv安装在不在路径上的'/usr/local/airflow/.local/bin中。

Answer 1

Airflow 使用 Shutil.which 来查找 virtualenv。通过requirements.txt安装的virtualenv不在路径上。将 virtualenv 的路径添加到 PATH 可以解决此问题。

这里的文档是错误的 https://docs.aws.amazon .com/mwaa/latest/userguide/samples-virtualenv.html

import os
from airflow.plugins_manager import AirflowPlugin
import airflow.utils.python_virtualenv 
from typing import List
def _generate_virtualenv_cmd(tmp_dir: str, python_bin: str, system_site_packages: bool) -> List[str]:
    cmd = ['python3','/usr/local/airflow/.local/lib/python3.7/site-packages/virtualenv', tmp_dir]
    if system_site_packages:
        cmd.append('--system-site-packages')
    if python_bin is not None:
        cmd.append(f'--python={python_bin}')
    return cmd
airflow.utils.python_virtualenv._generate_virtualenv_cmd=_generate_virtualenv_cmd
#This is the added path code
os.environ["PATH"] = f"/usr/local/airflow/.local/bin:{os.environ['PATH']}"
class VirtualPythonPlugin(AirflowPlugin):                
    name = 'virtual_python_plugin'

Answer 2

我想对公认的答案发表评论，但我还没有足够的声誉。

如果您处于MWAA的最新版本（使用气流2.4.3），那么您需要更改以下行：

cmd = ['python3','/usr/local/airflow/.local/lib/python3.7/site-packages/virtualenv', tmp_dir]

对此：

cmd = ['python3','/usr/local/airflow/.local/lib/python3.10/site-packages/virtualenv', tmp_dir]

要考虑Python的新版本