损坏的DAG：（...）没有名为docker的模块

Question

我有所有运行的BigQuery连接器，但是我希望在Cloud Composer而不是App Engine Flexible上计划在Docker容器中有一些现有脚本。

我有以下脚本似乎遵循了我可以找到的示例：

import datetime
from airflow import DAG
from airflow import models
from airflow.operators.docker_operator import DockerOperator

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())

default_args = {
    # Setting start date as yesterday starts the DAG immediately
    'start_date': yesterday,
    # If a task fails, retry it once after waiting at least 5 minutes
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
}

schedule_interval = '45 09 * * *'

dag = DAG('xxx-merge', default_args=default_args, schedule_interval=schedule_interval)

hfan = DockerOperator(
   task_id = 'hfan',
   image   = 'gcr.io/yyyyy/xxxx'
 )

...但是当尝试运行时，它在网络用户界面中告诉我：

Broken DAG: [/home/airflow/gcs/dags/xxxx.py] No module named docker

也许Docker未配置为在Cloud Composer运行的Kubernetes集群内工作吗？还是我只是缺少语法中的某些内容？

Answer 1

我通过在composer的PyPI部分中安装docker-py == 1.10.6来解决它。

但是，要使DockerOperator正常工作，需要付出更多的努力，因为作曲家的工作人员无法访问Docker守护程序。转到GCP控制台并执行以下步骤；获取群集凭据后）。

1. 将当前部署配置导出到文件

   kubectl get deployment airflow-worker -o yaml --export > airflow-worker-config.yaml

2. 编辑airflow-worker-config.yaml（示例链接）以挂载docker.sock和docker，授予对airflow-worker的特权访问以运行docker命令

3. 应用部署设置

   kubectl apply -f airflow-worker-config.yaml

Answer 2

这意味着：无论您的 Airflow 实例安装在何处，docker都缺少指定的 Python 包。

如果我配置我的个人机器，我可以安装缺少的包

pip install docker

编辑

在 docker 组件的源代码中https://airflow.incubator.apache.org/_modules/airflow/operators/docker_operator.html

有一个导入语句：

from docker import Client, tls

因此，cannot import name Client在我看来，新错误与安装损坏或docker软件包版本错误有关。

Answer 3

正如其他答案中所述，Docker Python 客户端未预安装在 Cloud Composer 环境中。要安装它，请将其添加为环境配置中的 PyPI 依赖项。

警告：默认情况下，DockerOperator将尝试与 Docker API 对话/var/run/docker.sock来管理容器。此套接字未安装在 Composer Airflow 工作单元内，因此不建议手动配置它。仅当配置为与在您的环境之外运行的 Docker 守护进程对话时，才建议在 Composer 中使用DockerOperator。

为了避免更脆弱的配置或绕过 Kubernetes 带来的意外（因为它负责管理整个集群中的容器），您应该使用KubernetesPodOperator. 如果您要将容器启动到 GKE 集群（或 Composer 环境的集群）中，则可以使用GKEPodOperator，它具有更具体的 GCP 相关参数。