问题标签 [dockeroperator]

我让 Airflow 在容器中运行。

我使用 Docker Operator 创建了许多运行脚本的容器。

容器的名称是使用一些参数动态创建的，如果我列出容器以通过终端查看发生了什么，这对我来说很有意义。

例如，容器名称称为 my_script_daily_A

如果脚本成功运行，容器将自动删除。

如果任务失败，我会重试。如果脚本失败并且气流重试运行，则会出现问题：

如果脚本失败，容器仍会以退出状态存在，不会被删除。这使得它的名称已经分配，​​因此当 Airflow 再次运行任务时，它会引发错误“具有此名称的容器已经存在”。

为了解决这个问题，我创建了一个每分钟在主机上运行的脚本：

我对此不太满意，我宁愿只有在失败时才触发这个脚本，如果我能检测到失败是否是由于名称冲突引起的就更好了。

我知道您可以使用“on_failure_callback”属性设置回调操作。

但是，只有在我直接在主机上而不是在容器上运行 Airflow 时，运行我的脚本来删除退出的容器才有效。

关于在这里做什么的任何建议？

当我第二次使用 dockeroperator 拉取相同的 docker 图像时，它正在从本地检索，但我希望 dockeroperator 每次都从 repo 中拉取图像，气流中是否可能？

我们有几个 DAG 处于活动状态的 Airflow 正在运行（使用 Docker compose）。上周我们将 Airflow 更新到了 2.1.3 版本。这导致我们使用 DockerOperator 的 DAG 出现错误：

我发现这个发行说明告诉我

气流.providers.docker.operators.docker.DockerOperator 和气流.providers.docker.operators.docker_swarm.DockerSwarmOperator 中的volumes 参数被mounts 参数替换

所以我把我们的 DAG 从

对此

但现在我得到这个错误：

我究竟做错了什么？

我正在尝试设置从 .bag 文件中提取图像的 Airflow ETL 管道。我想在 docker 中提取它，我正在使用 DockerOperator。Docker 镜像是从私有 GitLab 存储库中提取的。我要运行的脚本是 Docker 容器内的 python 脚本。.bag 文件在我的外部 SSD 上，所以我试图将它安装在 docker 中。代码有问题还是另一种问题？

错误：

这是我的代码：

我正在使用 docker-compose 和 celery 执行器运行 Airflow 2.1.4。到目前为止，我已经能够从 celery 工作容器启动和运行简单的 DockerOperator 任务，但是现在当我尝试将目录从共享驱动器挂载到任务容器时，出现错误（下面的日志文件）。如果我没有定义 mounts 参数，Dag 可以正常工作。所以我猜一些信息或特权没有传递给由 celery worker 容器创建的容器。有什么建议下一步该尝试什么吗？

DAG 文件：

日志文件：

我想做的是在 DockerOperator 中运行一个 docker 映像。在容器内，我想为 .py 文件启动一个 python 交互式 shell 并在文件中运行一个方法。

这可能在一个运营商中实现吗？

我的气流在一个 IP 上。我的 Docker 映像托管在另一个 IP 上。如何使用 docker 运算符从气流中安排我的 docker 映像。

`

`

我试图在我的 DAG 中将一个值从一个任务传递给另一个任务。

我的 DockerOperator 任务中有这段代码：

然而，我从上面的“ti.xcom_pull”这一行得到错误“无效的语法”。

这个 DAG 有两个任务，第一个是任务 ID run_this，它调用一个名为run_this_now的 Python 可调用函数，并将this 作为其返回语句：return payload

我正在查看 Admin-->XComs 下的 UI，并查看键 'return_value' 及其值，如任务 ID 'run_this' 返回的值。

所以我知道我有一个带有返回值的 XCom，但语法仍然给我这个错误。

我正在运行 Airflow 2.0.2