【🔒 MLOps - Airflow】使用DAG流程呼叫Docker Operator來執行任務

我們在「【🔒 MLOps - Airflow】使用docker compose架設Airflow」有教您如何使用Docker來架設Airflow的環境， 而這個章節我們準備在DAG的流程中設計一個呼叫Docker去執行作業的節點， 這對於我們容器化的工作流會非常有幫助， 而且運行過程中也減少對主機的干擾， 那就讓我們來好好的理解一番吧！

在進入主題之前先來個預告， 預計今天我們會學到這幾個部份：

下載官方compose檔

curl -LfO '<https://airflow.apache.org/docs/apache-airflow/2.10.0/docker-compose.yaml>'

首先我們寫一段DAG來叫Python的Container執行程式碼

這邊我們設計一個名為「docker_operator_example」的DAG， 很簡單的一個流程， 僅使用「python:3-alpine」這個Docker Image並執行python的指令。

from airflow import DAG
from airflow.providers.docker.operators.docker import DockerOperator
from airflow.utils.dates import days_ago

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
}

with DAG(
    dag_id='docker_operator_example',
    default_args=default_args,
    schedule_interval=None,
    catchup=False,
) as dag:

    run_docker = DockerOperator(
        task_id='run_docker',
        image='python:3-alpine',  # 要運行的 Docker 映像
        command='python -c \\'print("Hello, World!")\\'',  # 容器中要運行的命令
        docker_url='unix://var/run/docker.sock',  # Docker daemon 的 URL
        network_mode='bridge',  # Docker 網絡模式
        auto_remove=True,  # 在運行結束後自動移除容器
    )

    run_docker

接著我們把這個「docker_operator_example.py」放到./dags底下， 目錄大概會是這樣：

|- dags
	|- docker_operator_example.py
|- compose.yml

接著啟動服務並執行看看這個DAG

我們用以下指令來啟動服務：

docker compose up -d

打開 http://localhost:8080/home 並找到「docker_operator_example」的DAG， 並且在右方的按鈕執行它。

奇怪！ 執行完之後怎麼發生「Failed to establish connection to any given Docker hosts.」 這樣的錯誤訊息？

這個錯誤我們專門開設一篇「【MLOps - Airflow 疑難雜症】在Airflow Container裡面跑Docker Operator」來說明原因， 詳細原因請閱讀該文章。

如何讓Docker互通

由於我們的Airflow是在Container環境底下， 當DAG使用Docker Operator之後， 我們Container裡面要如何與外界溝通？ 這時候就得透過一個重要的 docket.sock 來進行了， 我們在「【MLOps - Airflow 疑難雜症】在Airflow Container裡面跑Docker Operator」分享到， 有興趣的朋友也歡迎閱讀我們的疑難排解過程， 裡面主要在講述DockerOperator來執行任務時會遇到的困難點及解決方案， 在這邊我們就不贅述了， 我們直接選擇docker proxy的方式來解這問題。

只要兩個步驟就能完成：

1. 加入docker-socket-proxy到compose。

2. DAG改用docker-socket-proy的連線位置。

run_docker = DockerOperator(
        task_id='run_docker',
        image='python:3-alpine',  # 要運行的 Docker 映像
        command='python -c \\'print("Hello, World!")\\'',  # 容器中要運行的命令
        docker_url='TCP://docker-socket-proxy:2375',  # Docker daemon 的 URL
        network_mode='bridge',  # Docker 網絡模式
        auto_remove=True,  # 在運行結束後自動移除容器
)

結語

由於我們的airflow架設於容器化的環境之下， 因此如果所有的任務都要用容器化來完成的話勢必會需要了解Docker In Docker的狀況， 而我們這邊就使用了Docker Proxy的代理方式來幫我們呼叫外部的容器完成任務， 當然還有更多方法可以使用， 也歡迎您與我們分享更多的解決方案， 讓我們一同交流學習！