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

Question

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

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

编辑:我希望使用固有的tfx AirflowDagRunner，它是基于python操作符的。我不希望更改任何tfx代码来处理这种情况。

Answer 1

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

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

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

Answer 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
)

Airflow文档：

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