[Airflow/AWS] Airflow - Trigger Rule

2023.05.25 (Thu) 학습 정리
#Linux-if문 #chmod #TriggerRule

---

1. Linux

1-1. If문

• 기본 사용법

if [ 조건식1 -a 조건식2 ]
then 
    수행문
elif [ 조건식 ]
    수행문
else
    수행문
fi

• 논리 연산자
  • and : -a
  • or : -o
• input 값 받아서 처리하기
  • $n : n번째 값
  • $# : input 값의 수

📌 if문 활용 예제

if [[ $# != 3 ]] #input값의 수가 3이 아닐 경우
then
  echo "Oops!!! argument required."
  exit 1
fi

echo $1
echo $2
echo $3

$ bash a.sh 1 2
Oops!!! argument required.
$ bash a.sh 1 2 3
1
2
3

1-2. chomod +x

디렉토리 및 파일에 실행 권한을 적용

chmod +x 파일명

---

2. Airflow Trigger Rule

상위 Task와의 의존관계를 설정하여 Task가 성공/실패 경우에 따라 다음 Task를 실행

all_success	모든 상위 Task 실행 성공
all_failed	모든 상위 Task 실행 실패
all_done	모든 상위 Task (성공/실패 여부 상관없이) 실행 완료
one_failed	상위 Task 중 1개 이상 실패 - 모든 상위 Task의 실행을 기다리지 않음
one_success	상위 Task 중 1개 이상 성공 - 모든 상위 Task의 실행을 기다리지 않음
none_failed	모든 상위 Task가 실패 또는 upstream_failed가 아님 (모든 상위 Task가 성공 또는 skipped)
none_failed_min_one_success	모든 상위 Task가 실패 또는 upstream_failed가 아니고 1개 이상 성공
none_skipped	모든 상위 Task가 skipped가 아님 (모든 상위 Task가 성공, 실패 또는 upstream_failed)
always	항상 실행

• Trigger Rule 설정 

test_bash = BashOperator(
        task_id = 'test',
        bash_command=f"echo '1'",
        dag = test_dag,
        trigger_rule="all_success", #trigger_rule 작성
        )

 

• Trigger Rule 활용

check.sh를 통해 경로에 해당 파일의 존재 여부를 체크하고 상황에 따라 다르게 task가 실행되도록 코드를 구현함 

 

📌 Trigger Rule 활용 예제 코드

#check.sh

if [[ $# != 2 ]]
then
  echo "Oops!!! hdfs_path argument required."
  exit 1
fi

S3_PATH=$1
DONE_FLAG_NAME=$2

aws s3 ls ${S3_PATH}/${DONE_FLAG_NAME}

if [ $? -eq 0 ]
then
    echo "confirmed : zero size ${DONE_FLAG_NAME}"
    exit 0
else
    echo "${DONE_FLAG_NAME} is missing or non-zero size"
    exit 1
fi

chmod +x [경로/파일명.sh] #권한부여

#dags.py

from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator

MY_PATH = '/opt/airflow/dags/sub'

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 5, 1),
    'retries': 0,
}
test_dag = DAG(
        'sub_test',
    default_args=default_args,
    schedule_interval=timedelta(days=1)
)
# Define the BashOperator task
check_task = BashOperator(
    task_id='Check_Done',
    bash_command=f"""
       {MY_PATH}/check.sh s3://sub DONE
    """,
    dag=test_dag
)

done_exit = BashOperator(
        task_id = 'Done_Exit',
        bash_command=f"echo 'confirmed : zero size _DONE'",
        dag = test_dag,
        trigger_rule="all_success",
        )
        
        
done_not_exit = BashOperator(
        task_id = 'Done_not_Exit',
        bash_command=f"echo 'DONE is missing or non-zero size'",
        dag = test_dag,
        trigger_rule="none_failed",
        )

check_task >> [done_exit, done_not_exit]