airflow 도커 오퍼레이터

회사에서 airflow를 도입했는데

도커 오퍼레이터를 쓰는게 그렇게 쉽지는 않아서

지금 도커 허브에 올라온게 2.5 버전이 올라 왔고

문서는 3.4 버전이 stable이라

이글을 볼 때 지금 설치 버전과 문서 버전을 꼭 확인해 보라!

sudo chmod 666 /var/run/docker.sock

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.docker_operator import DockerOperator
from airflow.operators.python_operator import BranchPythonOperator
from airflow.operators.dummy_operator import DummyOperator
from docker.types import Mount

default_args = {
'owner'                 : 'airflow',
'description'           : 'Use of the DockerOperator',
'depend_on_past'        : False,
'start_date'            : datetime(2021, 5, 1),
'email_on_failure'      : False,
'email_on_retry'        : False,
'retries'               : 1,
'retry_delay'           : timedelta(minutes=5)
}


with DAG('himari2', default_args=default_args, schedule_interval="0 */4 * * *", catchup=False) as dag:
    start_dag = DummyOperator(
        task_id='start_dag'
        )

    end_dag = DummyOperator(
        task_id='end_dag'
        )

    t1 = BashOperator(
        task_id='print_current_date',
        bash_command='date'
        )

    t2 = DockerOperator(
        task_id='himari_collector',
        image='centos:latest',
        api_version='auto',
        auto_remove=True,
        docker_url="unix://var/run/docker.sock",
        network_mode="host",
        environment={'ENVIRONMENT':'linux'},
        mount_tmp_dir=False,
        mounts=[
            Mount(source='/home/mx/docker/airflow/var/pos',
              target='/var/pos',
              type='bind'),
            Mount(source='/home/mx/docker/airflow/var/data',
              target='/var/data',
              type='bind'),
        ],
        )

    t4 = BashOperator(
        task_id='print_hello',
        bash_command='echo "hello world"'
        )

    start_dag >> t1

    t1 >> t2 >> t4

    t4 >> end_dag