[Airflow] docker run 명령어를 DockerOperator 로 구성해보기 (예시)

Airflow 에서는 Docker 를 사용하기위한 DockerOperator 가 존재한다.

그리고 도커 이미지가 잘 동작하는지 확인하기위해 docker run 명령어를 이용해서 실행해보고 이를 airflow 의 DAG 로 등록하는 경우가 많을것이라고 생각한다. 그런데 이걸 DockerOperator 로 구성할때 인자값이 헛갈리는데 이를 비교해서 구성하는 예시를 안내하로자 한다.

 

Docker 커맨드로 실행하는 예제

아래와 같은 명령어로 docker run 을 테스트해보았다고 가정하겠다.

볼륨 마운트와 EntryPoint 변경 그리고 CMD 에 환경변수 까지 존재하는 여러 케이스를 나열한 샘플이다.

그럼 이걸 AirflowDAG 로는 어떻게 표현할까?

% docker run --entrypoint /keytab/kerberos_init.sh --rm\
   -v /ext/keytab/:/keytab\
   -e USER_NAME=gildong@MYHOME.COM\
   -e KEYTAB_PATH=/keytab/gildong.keytab
   myhome.hahaha.com/hadoop-client:latest\
   hadoop fs -ls

 

kerberos_init.sh

#!/bin/bash
# ex: kinit -kt /keytab/gildong.keytab gildong@MYHOME.COM
kinit -kt ${KEYTAB_PATH} ${USERNAME}
$@

 

Airflow DAG 로 구성하기

위의 docker run 의 명령어을 Airflow 의 DAG 로 표현하면 아래와 같다. 참고로 아래 샘플은 airflow 2.x 버전대의 DAG 이다.

from airflow import DAG
from airflow.operators.docker_operator import DockerOperator
from datetime import datetime, timedelta
from docker.types import Mount 

default_args = {
    'owner'                 : 'gildong',
    'description'           : 'docker_sample_dag',
    'depend_on_past'        : False,
    'start_date'            : datetime(2022, 1, 1),
    'email_on_failure'      : True,
    'email_on_retry'        : False,
    'email'                 : 'gildong@myhome.com',
}
with DAG('docker_sample_dag', default_args=default_args, schedule_interval=None, catchup=True) as dag:
    t = docker_operator = DockerOperator(
       task_id='docker_sample_task',
       image='myhome.hahaha.com/hadoop-client:latest',
       entrypoint="/keytab/kerberos_init.sh",
       command="hadoop fs -ls",
       environment={"USER_NAME": "gildong@MYHOME.COM", "KEYTAB_PATH": "/keytab/gildong.keytab"},
       mounts=[Mount(source='/ext/keytab', target='/keytab', type='bind')],
       api_version='auto',
       auto_remove=True,
       docker_url="unix://var/run/docker.sock",
       network_mode="bridge"
    )
    t

여기서 주의할점은 어떤 Worker 서버에서 실행될지 모르기 때문에 "/ext/keytab" 경로 같은 로컬경로의 파일은 모든서버에서 접근이 가능한 공유 스토리지여야 한다.  (ceph 스토리지를 마운트 할수도 있고 NFS 를 마운트 할수도 있고 이건 구성하기 나름일듯 하다)

이런 스토리지 문제의 경우는 더 좋은 방법이 있을거 같긴한데 이부분은 나중에 더 좋은 방법을 찾으면 포스팅하도록 하겠다.

 

좀 더 다양한 인자값의 설명은 airflow 홈페이지의 document 를 참고하도록 하자.

https://airflow.apache.org/docs/apache-airflow-providers-docker/2.7.0/_api/airflow/providers/docker/operators/docker/index.html

airflow.providers.docker.operators.docker — apache-airflow-providers-docker Documentation