气流-在run_id中使用DockerOperator作为卷名

Question

我用Apache气流写了一个DockerOperator，我想给它一个卷。到目前一切尚好。下面是一个示例：

t = DockerOperator(
        task_id='test',
        image='testimage:latest',
        command='python3 /code/test.py',
        volumes=["/mnt/interim:/interim"],
        xcom_push=True,
        dag=dag,
)

我的问题是：

挂载目录的名称需要灵活。因此，我想挂载一个名为run_id的目录。

volumes=["/mnt/interim/" + "{{ run_id }}" + ":/interim"]

然而，气流似乎并没有在卷中解析"{{ run_id }}“，而只在DockerOperator的命令中解析。

简单地说，我想获得run_id以便挂载它。

请注意，使用气流变量(气流的环境变量)并不能做到这一点，因为如果任务并行运行，这个变量可能会被覆盖。

也许你们中的一些人已经知道一个高级的DockerOperator可以做到这一点(CustomOperator)。

(预先谢谢:)

Answer 1

谢谢约翰斯贴出这个问题。

您正在尝试实现的是可能的，但是因为这不是一个非常常见的用例，所以默认情况下不会启用。template_fields迭代中的参数是由气流模版的。volumes字段不在那里，因此不会被拾取。

最简单的方法是在项目中复制docker_operator.py，并将volumes字段添加到列表中：https://github.com/apache/airflow/blob/master/airflow/operators/docker_operator.py#L126

template_fields = ('command', 'environment', 'container_name', 'volumes',)

您也可以打开一张票证并将其合并到流中，但我不确定有多少用户将模板这个字段。希望这能有所帮助。