Airflow：如何使用 tfx 的 AirflowDagRunner 在运行时为 tfx 管道创建分布式气流中的虚拟 python 环境？答案

【问题标题】：Airflow: How can I create a virtual python environment in distributed airflow in runtime for a tfx-pipeline using the tfx's AirflowDagRunner?Airflow：如何使用 tfx 的 AirflowDagRunner 在运行时为 tfx 管道创建分布式气流中的虚拟 python 环境？
【发布时间】：2021-07-01 11:48:40
【问题描述】：

我有 DAGS，我想在同一个分布式气流集群的不同 python 环境中运行。是否可以在执行 DAG 的同时创建虚拟 python 环境？

我希望将 tfx ML 管道添加到我们的分布式气流平台，并且无法在每个工作节点中单独安装 tfx。

编辑：我希望使用基于 python 运算符的固有 tfx AirflowDagRunner。我宁愿不更改任何 tfx 代码来处理此案。

【问题讨论】：

标签： tfx airflow

【解决方案1】：

您可以随意创建和拆除虚拟环境 (venv)。如果您的软件在被拆除之前没有等待环境的结果，那将是最好的。

要以编程方式创建环境，只需使用您喜欢的编程语言运行 shell 命令并传入常用命令行即可。如果您使用 venv 中的python 程序执行 Python 程序，则会选择 venv。

例如，如果您在 /var/venvs/123 中创建一个 venv，请使用类似 /var/venvs/123/bin/python 的命令行运行您的 Python 程序，并传递任何必要的参数。

【讨论】：

【解决方案2】：

你应该试试 PythonVirtualenvOperator，Airflow 会根据操作符参数创建特殊的 venv

带有 openpyxl 和 pandas 库的示例 venv：

extract_some_data = PythonVirtualenvOperator(
    task_id="extract_some_data ",
    python_callable=extract_some_data ,
    requirements=["openpyxl", "pandas"],
    python_version='3.7',
    system_site_packages=False,
    provide_context=True,
    op_kwargs={
        "uuid": "{{ ti.xcom_pull(key=None, task_ids='generate_uuid') }}"
    },
    dag=dag
)

气流文档：

https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/operators/python/index.html?highlight=pythonoperator#airflow.operators.python.PythonOperator

【讨论】：

如果可能的话，我想通过 ttx 的 AirflowDagRunner 来实现。 TFX 有一个 dagrunner，它使用 PythonOperator 的子类 AirflowComponents 执行每个节点。你知道我怎么用它吗？感谢您的帮助！