[Airflow] Bash Operator & 외부 쉘파일 수행하기

1. 쉘(Shell) 스크립트란?

• Unix/Linux Shell 명령을 이용하여 만들어지고 인터프리터에 의해 한 줄씩 처리되는 파일
• echo, mkdir, cd, cp, tar, touch 등의 기본적인 쉘 명령어를 입력하여 작성하며
  변수를 입력받거나 for문, if문 그리고 함수도 사용 가능
• 확장자가 없어도 동작하지만 주로 파일명에 .sh 확장자를 붙임

 

2. 왜 쉘(Shell) 스크립트가 필요한가?

• 쉘 명령어를 이용하여 복잡한 로직을 처리하는 경우
  ex) sftp를 통해 파일을 받은 후 DB에 Insert&tar.gz으로 압축해두기
• 쉘 명령어 재사용을 위해

 

3. Worker 컨테이너가 쉘 스크립트를 수행하려면?

• 문제점
  1) 컨테이너는 외부의 파일을 인식할 수 없음
  2) 컨테이너 안에 파일을 만들어주면 컨테이너 재시작시 파일이 사라짐
• 해결방법
  plugins 폴더에 쉘 파일을 갖다 놓으면 됨 (python 파일도 마찬가지)

 

 

 

4. Shell 스크립트 예제코드

아래 sh 파일을 로컬 /airflow/plugins/shell 에 생성합니다.

(yaml 파일에서 로컬의 plugins 경로를 /airflow/plugins로 변경해야 합니다.)

/*select_fruit.sh*/
FRUIT=$1
if [ $FRUIT == APPLE ];then
        echo "You selected Apple!"
elif [ $FRUIT == ORANGE ];then
        echo "You selected Orange!"
elif [ $FRUIT == GRAPE ];then
        echo "You selected Grape!"
else
        echo "You selected other Fruit!"
fi

 

 

4. Shell 스크립트 실행하는 bash 오퍼레이터 코드 예제

아래 py 파일을 로컬 /airflow/dags 에 생성합니다.

/*dags_bash_select_fruit.py*/
from airflow import DAG
import pendulum
import datetime
from airflow.operators.bash import BashOperator


with DAG(
    dag_id = "dags_bash_select_fruit",
    schedule = "10 0 * * 6#1",
    start_date = pendulum.datetime(2023, 3, 1, tz = "Asia/Seoul"),
    catchup = False
) as dag:
    
    t1_orange = BashOperator(
        task_id = "t1_orange",
        bash_command = "/opt/airflow/plugins/shell/select_fruit.sh ORANGE",
    )

    t2_avocado = BashOperator(
        task_id = "t2_avocado",
        bash_command = "/opt/airflow/plugins/shell/select_fruit.sh AVOCADO",
    )

    t1_orange >> t2_avocado

 

실행하면 아래와 같이 동작합니다.