我已经在本地成功开发了一个 super 简单的 ETL 流程(下面称为 load_staging),该流程从某个远程位置提取数据,然后将未处理的数据写入本地 Windows 计算机上的 MongoDB 容器中。现在,我想使用 DockerOperator 为每个任务使用 Apache-Airflow 来安排此过程,即我想创建源代码的 docker 镜像,然后使用 DockerOperator 执行该镜像中的源代码。由于我在 Windows 机器上工作,因此我只能从 Docker 容器内部使用 Airflow。
我已经使用docker-compose up
启动了airflow容器(下面称为webserver)和MongoDB容器(下面称为mongo),并在Airflow的GUI中手动触发了DAG。根据 Airflow 的说法,任务正在成功执行,但似乎 docker 镜像内的代码没有被执行,因为任务完成得太快,并且在从我的镜像启动 docker 容器后,任务执行时出现错误代码 0,即我没有看到任务本身的任何日志输出。请参阅下面的日志:
[2020-01-20 17:09:44,444] {{docker_operator.py:194}} INFO - Starting docker container from image myaccount/myrepo:load_staging_op
[2020-01-20 17:09:50,473] {{logging_mixin.py:95}} INFO - [[34m2020-01-20 17:09:50,472[0m] {{[34mlocal_task_job.py:[0m105}} INFO[0m - Task exited with return code 0[0m
所以,我的两个问题是:
- 我是否得出了正确的结论,或者这个问题的根源可能是什么?
- 如何确保图像内的代码始终被执行?
下面您可以找到有关如何设置 DockerOperator、如何定义应该由 DockerOperator 执行的镜像、启动 Web 服务器的 docker-compose.yml 文件以及mongo 容器和用于创建 Web 服务器容器的 Dockerfile。
在我的 DAG 定义文件中,我指定了 DockerOperator,如下所示:
CONFIG_FILEPATH = "/configs/docker_execution.ini"
data_object_name = "some_name"
task_id_ = "{}_task".format(data_object_name)
cmd = "python /src/etl/load_staging_op/main.py --config_filepath={} --data_object_name={}".format(CONFIG_FILEPATH, data_object_name)
staging_op = DockerOperator(
command=cmd,
task_id=task_id_,
image="myaccount/myrepo:load_staging_op",
api_version="auto",
auto_remove=True
)
上面引用的镜像 load_staging_op
的 Dockerfile 如下所示:
# Inherit from Python image
FROM python:3.7
# Install environment
USER root
COPY ./src/etl/load_staging_op/requirements.txt ./
RUN pip install -r requirements.txt
# Copy source code files into container
COPY ./configs /configs
COPY ./wsdl /wsdl
COPY ./src/all_constants.py /src/all_constants.py
COPY ./src/etl/load_staging_op/utils.py /src/etl/load_staging_op/utils.py
COPY ./src/etl/load_staging_op/main.py /src/etl/load_staging_op/main.py
# Extend python path so that custom modules are found
ENV PYTHONPATH "${PYTHONPATH}:/src"
ENTRYPOINT [ "sh", "-c"]
docker-compose.yml
文件的相关方面如下:
version: '2.1'
services:
webserver:
build: ./docker-airflow
restart: always
privileged: true
depends_on:
- mongo
- mongo-express
volumes:
- ./docker-airflow/dags:/usr/local/airflow/dags
# source code volume
- ./src:/src
- ./docker-airflow/workdir:/home/workdir
# Mount the docker socket from the host (currently my laptop) into the webserver container
# so that we can build docker images from inside the webserver container.
- //var/run/docker.sock:/var/run/docker.sock # the two "//" are needed for windows OS
- ./configs:/configs
- ./wsdl:/wsdl
ports:
# Change port to 8081 to avoid Jupyter conflicts
- 8081:8080
command: webserver
healthcheck:
test: ["CMD-SHELL", "[ -f /usr/local/airflow/airflow-webserver.pid ]"]
interval: 30s
timeout: 30s
retries: 3
networks:
- mynet
mongo:
container_name: mymongo
image: mongo
restart: always
ports:
- 27017:27017
networks:
- mynet
上述 Dockerfile 中引用的 Web 服务器容器的 Dockerfile 如下所示:
FROM puckel/docker-airflow:1.10.4
# Adds DAG folder to the PATH
ENV PYTHONPATH "${PYTHONPATH}:/src:/usr/local/airflow/dags"
# Install the optional packages
COPY requirements.txt requirements.txt # make sure something like docker==4.1.0 is in this requirements.txt file!
USER root
RUN pip install -r requirements.txt
# Install docker inside the webserver container
RUN curl -sSL https://get.docker.com/ | sh
ENV SHARE_DIR /usr/local/share
# Install simple text editor for debugging
RUN ["apt-get", "update"]
RUN ["apt-get", "-y", "install", "vim"]
感谢您的帮助,非常感谢!
最佳答案
衷心感谢所有花时间帮助我解决问题的人。我需要实现以下更改才能使其正常工作:
Docker 操作符:
- 调整在运行时(即构建容器时)传递给容器的命令
- 将参数
network_mode
添加到 Web 服务器容器运行的网络。这对我来说很困难,因为我是 Docker 新手,在网上找不到太多关于此的教程。为了查找运行 Web 服务器容器的网络名称,我使用诸如 docker network ls 之类的工具列出了主机(=windows 笔记本电脑)上当前所有事件的网络。在显示的网络列表中,我看到一个名为project_root_dirname_mynet
的网络,它是我的项目根目录和 docker-compose.yml 中指定的网络名称的组合> 文件。有趣的是(而且显然),列出所有网络后,您可以使用诸如 docker network Inspection project_root_dirname_mynet 之类的命令来检查网络project_root_dirname_mynet。这将返回一个带有“containers”小节的 json 文件,您可以在其中看到 docker-compose.yml 文件中指定的所有容器。
DockerOperator 的代码变为:
cmd = "--config_filepath {} --data_object_name {}".format(CONFIG_FILEPATH.strip(), data_object_name.strip())
print("Command: {}".format(cmd))
staging_op = DockerOperator(
command=cmd,
task_id=task_id_,
image="myaccount/myrepo:load_staging_op",
api_version="auto",
auto_remove=True,
network_mode="project_root_dirname_mynet"
)
load_staging_op 任务的 Dockerfile:
- 将最后一行从
ENTRYPOINT [ "sh", "-c"]
更改为ENTRYPOINT [ "python", "/src/etl/load_staging_op/main.py"]
。我认为“python”参数将在容器中打开一个Python控制台,第二个参数只是您想要在docker容器内执行的脚本的路径。然后,在运行时(或构建时或无论如何调用),上面的cmd
中的命令行参数将被传递。在图像的源代码中,您可以使用像 argparse 这样的库来检索这些命令。
关于python - 如何在 Windows 主机上从 apache Airflow 使用 DockerOperator,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/59828386/