带有需要 2.7 的运算符的 Python 3.6 Airflow

Question

我目前正在 python 3.6.5 上运行 airflow (1.9.0) 实例。我有一个手动工作流程，我想将其移至 DAG。这个手动工作流现在需要用 python 2 和 3 编写的代码。让我们将我的 DAG 简化为 3 个步骤:

1. 处理数据并为机器设置数据的数据流作业学习培训
2. Tensorflow ML 训练作业
3. 其他我使用 python 3 代码编写的 PythonOperators

数据流作业是用 python 2.7 编写的(google 要求)，tensorflow 模型代码是用 python 3 编写的。查看 airflow 1.9.0 中的“MLEngineTrainingOperator”，有一个 python_version 参数设置“使用的 Python 版本”在训练中。

问题:

• 我可以在 worker 中动态指定特定的 python 版本吗环境？
• 我是否必须在 python 2.7 上安装 airflow 才能运行第 1) 步？
• 我可以在 python 3 中使用 tensorflow 模型代码，这些代码只是通过在 python 2 上运行的 MlEngineTraining 打包并提交吗？
• 我是否必须在 python 2 中重写我的 3) 运算符？

Answer 1

没有办法在 worker 上动态指定 python 版本。但是，如果您使用的是 Celery 执行程序，则可以在不同的服务器/虚拟机或不同的虚拟环境中运行多个工作程序。

您可以让一名工作人员运行 python 3，另一名工作人员运行 2.7，并让每个工作人员监听不同的队列。这可以通过三种不同的方式完成:

• 启动 worker 时，您可以添加一个 -q [queue-name] 标志
• 设置AIRFLOW__CELERY__DEFAULT_QUEUE的环境
• 更新 airflow.cfg 中 [celery] 下的 default_queue。

然后在您的任务定义中指定一个 queue 参数，根据任务需要运行的 python 版本更改队列。

我不熟悉 MLEngineOperator，但您可以在 PythonOperator 中指定 python_version，它应该在该版本的 virtualenv 中运行。或者，您可以使用 BashOperator，编写要在不同文件中运行的代码，并指定 python 命令以使用您要使用的 python 版本的绝对路径运行它。

无论任务如何运行，您只需要确保 DAG 本身与您运行它的 python 版本兼容。 IE。如果您要在不同的 python 版本中启动 airflow worker，则 DAG 文件本身需要与 python 2 & 3 兼容。 DAG 可以具有它所使用的具有版本不兼容性的附加文件依赖项。