标签: dockeroperator

我对 Docker + Airflow 很陌生。以下是我正在努力完成的工作。

我有 4 个服务，如下面的撰写文件所示。3 个与 Airflow 相关，一个作为测试 Ubuntu 实例。Airflow 相关容器：airflow-database, airflow-webserver,airflow-scheduler能够相互通信，我能够运行示例 DAG。现在我添加了第 4 个服务 (ubuntu)，我尝试使用 DockerOperator 从 DAG 向其发送一个简单的命令“/bin/sleep 10”（下面是 DAG 文件）。但由于某种原因，我收到了 Permission Denied 消息（还附上了 DAG 错误文件）。

如果我从本地主机运行 Airflow 而不是从 docker 容器内部运行 Airflow，它会起作用无法弄清楚我缺少什么。以下是我尝试过的一些方法：

• 在 docker_url 中：

1. 替换unix://var/run/docker.sock为tcp://172.20.0.1认为它可以通过 docker 主机 ip 解析

2. 使用 gateway.host.internal

3. 甚至从操作员中删除了 docker_url 选项，但意识到它无论如何都会默认为 unix://var/run/docker.sock

4. 尝试了一堆组合，tcp://172.20.0.1:2376, tcp://172.20.0.1:2375

5. 将主机端口映射到 Ubuntu，如 8085:8085 等。

• 可能是 Airflow 网络服务器的气流用户被 Ubuntu 踢出
• 因此，在 Ubuntu 容器中创建了一个组并向其添加了气流用户 - 不行
• api_version：选项“自动”也不起作用，并不断给出版本未找到错误。因此，我在docker version命令中发现了这一点，因此不得不使用 …

我正在尝试将本地目录作为卷传递给气流，而气流又被传递给 dag DockerOperator

我的airflow-docker-compose.yaml（在airflow-common部分）如下所示

  volumes:\n    - ./dags:/opt/airflow/dags\n    - ./logs:/opt/airflow/logs\n    - ./plugins:/opt/airflow/plugins\n    - ./input:/opt/airflow/input\n    - ./output:/opt/airflow/output\n

在 DAG 代码中，我尝试传递安装参数，如下所示：

eod_price = DockerOperator(\n        task_id='run_docker',\n        image='alpine',\n        api_version='auto',\n        command='/bin/touch /output/run_docker_touch.txt',\n        auto_remove=True,\n        mounts=[\n            Mount(source='/opt/airflow/output',\n                  target='/app_base/output',\n                  type='volume'),\n        ],\n        mount_tmp_dir=False,\n        docker_url='tcp://docker-proxy:2375',\n        network_mode='bridge'\n    )\n

使用这段代码，我收到错误：

\n

错误请求 ("create /opt/airflow/output: "/opt/airflow/output"\n仅包含本地卷名称的无效字符\n"[a-zA-Z0-9][a-zA-Z0- 9_.-]" 是允许的。如果您打算传递\n主机目录，请使用绝对路径")

\n

当我将安装行从 type='volume' 更改为 type='bind' 时：

mounts=[\n            Mount(source='/opt/airflow/output',\n                  target='/app_base/output',\n                  type='bind'),\n        ],\n

错误更改为

\n

错误请求（“类型“bind”的安装配置无效：绑定源路径\n不存在：/opt/airflow/output”）

\n

我猛击了 docker007-airflow-scheduler-1 、 docker007-airflow-triggerer-1 、 docker007-airflow-webserver-1 、 docker007-airflow-worker-1 ，在每个容器中我看到 /opt/airflow/output …

我们有 Airflow 运行（使用 Docker compose），并且有多个 DAG 处于活动状态。上周我们将 Airflow 更新至版本 2.1.3。这导致我们使用 DockerOperator 的 DAG 出现错误：

airflow.exceptions.AirflowException: Invalid arguments were passed to DockerOperator (task_id: t_docker). Invalid arguments were:
**kwargs: {'volumes':

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

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

所以我将 DAG 更改为

t_docker = DockerOperator(
        task_id='t_docker',
        image='customimage:latest',
        container_name='custom_1',
        api_version='auto',
        auto_remove=True,
        volumes=['/home/airflow/scripts:/opt/airflow/scripts','/home/airflow/data:/opt/airflow/data'],
        docker_url='unix://var/run/docker.sock',
        network_mode='bridge',
        dag=dag
    )

对此

t_docker = DockerOperator(
        task_id='t_docker',
        image='customimage:latest',
        container_name='custom_1',
        api_version='auto',
        auto_remove=True,
        mounts=['/home/airflow/scripts:/opt/airflow/scripts','/home/airflow/data:/opt/airflow/data'],
        docker_url='unix://var/run/docker.sock',
        network_mode='bridge',
        dag=dag
    )

但现在我得到这个错误：

docker.errors.APIError: 500 Server Error for http+docker://localhost/v1.41/containers/create?name=custom_1: Internal Server Error ("json: cannot unmarshal string into Go struct field …

我正在探索如何在 Apache Airflow 教程中使用 DockerOperator。我已成功使用 docker-compose 设置气流，并能够docker_dag在气流浏览器中访问教程中提到的内容。这是相同的代码。

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
from airflow.operators.docker_operator import DockerOperator
default_args = {
'owner'                 : 'airflow',
'description'           : 'Use of the DockerOperator',
'depend_on_past'        : False,
'start_date'            : datetime(2018, 1, 3),
'email_on_failure'      : False,
'email_on_retry'        : False,
'retries'               : 1,
'retry_delay'           : timedelta(minutes=5)
}
with DAG('docker_dag', default_args=default_args, schedule_interval="5 * * * *", catchup=False) as dag:
    t1 = BashOperator(
    task_id='print_current_date',
    bash_command='date'
    )
    t2 = …