Airflow(2)

이전 글

https://koreatstm.tistory.com/268

Airflow

실행(Execution or Trigger)

• DAG를 실제로 즉시 실행하거나, 예약된 스케줄에 따라 실행되는 행위
• 직접 UI에서 수동 실행(Trigger)할 수도 있고, 스케줄러에 의해 자동 실행될 수도 있음
• DAG가 실행되면 해당 DAG Run이 만들어짐
• 실행의 종류:  수동 실행(트리거), 예약 실행(스케줄)

 

스케줄(Schedule)

• DAG가 자동으로 언제 실행될지 주기를 설정하는 것
• 예: 매일 자정에 실행, 5분마다 실행, 매주 월요일 실행 등
• Airflow 스케줄러가 이 주기에 따라 DAG 실행을 예약하고 자동으로 실행시킴

 

궁금한 것!

• Q: task1 >> task2 >> task3 이렇게 task간 의존 관계 설정 안하면 순서가 어떻게 될까?
• 명시적으로 의존 관계를 지정하지 않으면, Airflow는 해당 태스크들을 병렬(독립) 작업으로 인식하고 실행 순서를 알아서 정한다.
• 이 경우 실행 순서는 스케줄러와 워커의 상태, 리소스, 우선순위 등에 따라 달라진다. -> 예측 불가능한 실행 순서

 

BashOperator

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
'''
DAG 자체 정의
컨텍스트 블록으로, Task들이 이 DAG의 일부라는 것을 명시
'''
with DAG(
    dag_id='every_10_seconds',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(seconds=10),
    catchup=False,
    max_active_runs=1,
    tags=['test'] #tag는 필터링을 위한 라벨

    
) as dag:
    

    task1 = BashOperator(
        task_id='print_time',
        bash_command='date "+%Y-%m-%d %H:%M:%S"'
    )

    task2 = BashOperator(
        task_id='say_hello',
        bash_command='echo "Hello Airflow!"'
    )

    task1 >> task2  # task1 실행 후 task2 실행
'''
DAG에 속한 Task 정의
DAG는 여러 Task로 구성된다.
'''

• catchup: Airflow는 start_date부터 현재 시간까지의 모든 실행 날짜에 대해 DAG을 실행하려고 한다.
• catchup=False : 과거 실행이 밀렸을 때 backlog를 따라잡지 않고 오늘부터 실행(오늘 이후만 실행하고, 과거 날짜는 무시한다.)

 

PythonOperator

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

# Python 함수 정의
def my_python_function():
    print("👋 Hello from PythonOperator!")

# DAG 정의
with DAG(
    dag_id='every_10_seconds_with_python',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(seconds=10),
    catchup=False,
    max_active_runs=1,
    tags=['test']
) as dag:

    # BashOperator Task
    print_time = BashOperator(
        task_id='print_time',
        bash_command='date "+%Y-%m-%d %H:%M:%S"'
    )

    # PythonOperator Task
    python_hello = PythonOperator(
        task_id='python_hello',
        python_callable=my_python_function
    )

    # Task 의존성 설정
    print_time >> python_hello

 

• DAG 실행 주기: 10초마다 한 번씩 실행
• Task 구성
  1. print_time (BashOperator) → 현재 시간을 출력 (date "+%Y-%m-%d %H:%M:%S")
  2. python_hello (PythonOperator) → my_python_function() 함수 실행 (프린트 메시지 출력)
• 의존성: print_time >> python_hello
• 즉, print_time이 성공적으로 끝나야만 python_hello가 실행됨

 

TaskGroup

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.task_group import TaskGroup
from datetime import datetime, timedelta

with DAG(
    dag_id='taskgroup_example',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(seconds=30),
    catchup=False,
    max_active_runs=1,
    tags=['group', 'example']
) as dag:

    start = BashOperator(
        task_id='start',
        bash_command='echo "Start DAG Execution"'
    )

    with TaskGroup("group1", tooltip="First Task Group") as group1:
        t1 = BashOperator(
            task_id='task_1_1',
            bash_command='echo "Group1 - Task 1"'
        )

        t2 = BashOperator(
            task_id='task_1_2',
            bash_command='echo "Group1 - Task 2"'
        )

    with TaskGroup("group2", tooltip="Second Task Group") as group2:
        t3 = BashOperator(
            task_id='task_2_1',
            bash_command='echo "Group2 - Task 1"'
        )

        t4 = BashOperator(
            task_id='task_2_2',
            bash_command='echo "Group2 - Task 2"'
        )

    end = BashOperator(
        task_id='end',
        bash_command='echo "End DAG Execution"'
    )

    # DAG 실행 흐름 정의
    start >> group1 >> group2 >> end

 

• TaskGroup은 단순히 Task들을 그룹으로 묶는 컨테이너이다.
• 내부 Task들은 병렬로 실행됨. (별도 의존관계 없을 경우)
• 즉, >>와 달리 그룹 내의 task들은 병렬로 실행된다. 다시말해, 그룹내의 task가 안되더라도, 그것과 상관없이 task는 진행된다.
• TaskGroup은 단순히 논리적인 그룹화를 제공하지만, 실제로 그룹 내의 Task들은 병렬로 실행된다.

 

XCom

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def push_function(**kwargs):
    # XCom에 값 보내기 (push)
    ti = kwargs['ti']
    ti.xcom_push(key='sample_key', value='Hello from push_function!')


def pull_function(**kwargs):
    # XCom에서 값 가져오기 (pull)
    ti = kwargs['ti']
    pulled_value = ti.xcom_pull(key='sample_key', task_ids='push_task')
    print(f"Pulled value from XCom: {pulled_value}")

with DAG(
    dag_id='xcom_example',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(seconds=30),
    catchup=False
) as dag:

    push_task = PythonOperator(
        task_id='push_task',
        python_callable=push_function,
    )

    pull_task = PythonOperator(
        task_id='pull_task',
        python_callable=pull_function,
    )

    push_task >> pull_task

 

 

• kwargs는 Airflow가 자동으로 전달하는 컨텍스트 정보 담음
• 이중에서 ti(TaskInstance객체)를 사용
• Airflow 내부 임시 저장소 XCom에 문자열 저장 됨

 

맨 밑 줄에 print문 확인 가능

 

Branch

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.empty import EmptyOperator 
from datetime import datetime, timedelta
import pendulum

# 분기 함수
def decide_what_to_do():
    today = pendulum.now("Asia/Seoul").day_of_week  # 0: Monday ~ 6: Sunday
    if today == 6:  # Sunday
        return 'take_rest'
    else:
        return 'work_hard'

with DAG(
    dag_id='branching_example',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(minutes=1),
    catchup=False,
    max_active_runs=1, # 해당 DAG 인스턴스는 하나씩만 실행 됨
    tags=['branch', 'example'],
) as dag:

    start = EmptyOperator(task_id='start')

    branch = BranchPythonOperator(
        task_id='branch_decision',
        python_callable=decide_what_to_do,
    )

    rest = EmptyOperator(task_id='take_rest')
    work = EmptyOperator(task_id='work_hard')

    # trigger_rule='none_failed_min_one_success: 이전 작업들 중 하나라도 성공하면 실행 됨
    end = EmptyOperator(task_id='end', trigger_rule='none_failed_min_one_success') 

    # DAG 흐름 정의
    start >> branch
    branch >> [rest, work] >> end

오늘이 일요일이므로 take_rest로 분기 됨

 

TaskGroup 중첩

 

SubDAG

• 기존 DAG 안에 또 다른 DAG를 넣는 방식
• 독립적인 DAG를 하나의 Task처럼 사용하는 개념
• 다만 Airflow에서는 SubDAG은 복잡성 때문에 권장하지 않는 추세이고, 대신 TaskGroup을 주로 사용

 

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.utils.task_group import TaskGroup
from datetime import datetime, timedelta

with DAG(
    dag_id='nested_taskgroup_example',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(minutes=5),
    catchup=False,
    max_active_runs=1,
    tags=['example', 'taskgroup'],
) as dag:

    start = EmptyOperator(task_id='start')

    # 최상위 TaskGroup
    with TaskGroup('group1') as group1:
        
        task_a = EmptyOperator(task_id='task_a')
        task_b = EmptyOperator(task_id='task_b')

        # 중첩된 TaskGroup
        with TaskGroup('subgroup1') as subgroup1:
            task_c = EmptyOperator(task_id='task_c')
            task_d = EmptyOperator(task_id='task_d')

        # 그룹 내부에서 subgroup1을 포함하여 작업 흐름 지정 가능
        task_a >> subgroup1 >> task_b

    end = EmptyOperator(task_id='end')

    start >> group1 >> end

 

• DAG 시작 → start task 실행
• group1 TaskGroup 실행
• task_a 실행 후, 중첩 TaskGroup인 subgroup1 실행 (task_c와 task_d 포함)
• 다시 task_b 실행
• group1 종료 후 end task 실행

 

 

Sensor

 

Sensor는 Airflow에서 어떤 조건이 충족될 때까지 기다리는 Task

• 파일이 도착할 때까지 기다리기 → FileSensor
• API 응답이 특정 값을 반환할 때까지 → HttpSensor
• 외부 테이블이 생길 때까지 → ExternalTaskSensor, SqlSensor

 

작동방식

모드	설명
poke	계속 실행 중인 상태로 대기하며 일정 주기마다 조건을 체크
reschedule	실행 중이 아니라 쉬었다가 다시 실행 (따라서, 자원을 적게 씀)

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime, timedelta

with DAG(
    dag_id='sensor_example',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(minutes=5),
    catchup=False,
    max_active_runs=1,
    tags=['sensor', 'example'],
) as dag:

    start = EmptyOperator(task_id='start')

    wait_for_file = FileSensor(
        task_id='wait_for_file',
        filepath='/opt/airflow/dags/input_file.txt',  
        poke_interval=10,     # 10초마다 확인
        timeout=60,           # 60초 동안 기다림 (없으면 실패)
        mode='poke',          # poke 모드: 주기적으로 확인
    )

    done = EmptyOperator(task_id='done')

    start >> wait_for_file >> done

최초 실행시에는 해당 파일이 없기에 에러가 뜬다. 터미널에서 아래 명령어로 file을 만들고, trigger해보면

 echo "test" > ./dags/input_file.txt

 

TriggerDagRunOperator 

 

현재 DAG에서 다른 DAG을 실행(trigger)

1. dags/ 폴더에 두 파일 저장
2. Airflow UI에서 parent_dag 실행
3. child_dag이 자동으로 트리거되는지 확인

 

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from datetime import datetime

with DAG(
    dag_id='child_dag',
    start_date=datetime(2025, 6, 29),
    schedule=None,
    catchup=False,
    tags=['TriggerDagRunOperator'],
) as dag:
    start = EmptyOperator(task_id='child_start')
    end = EmptyOperator(task_id='child_end')

    start >> end

# 8-2
from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from datetime import datetime, timedelta

with DAG(
    dag_id='parent_dag',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(minutes=5),
    catchup=False,
    tags=['TriggerDagRunOperator'],
) as dag:
    start = EmptyOperator(task_id='start')

    trigger_child = TriggerDagRunOperator(
        task_id='trigger_child_dag',
        trigger_dag_id='child_dag',
    )

    end = EmptyOperator(task_id='end')

    start >> trigger_child >> end

child_dag를 활성화해 놓고 parent_dag를 활성화하면, child_dag를 별도로 스케줄하지 않아도 parent_dag에서 TriggerDagRunOperator로 트리거되어 실행되는 것을 확인할 수 있다.

 

Variable

 

Airflow의 전역 설정값 저장소인 Variable을 사용해서 DAG 안에서 동적으로 값을 가져오기

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
from datetime import datetime, timedelta

def print_variable():
    filename = Variable.get("filename", default_var="default.txt")  # key=filename 없으면 default.txt 기본값
    print(f"📂 Airflow Variable 'filename' is: {filename}")

with DAG(
    dag_id='variable_example',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(minutes=10),
    catchup=False,
    tags=['variable'],
) as dag:

    task = PythonOperator(
        task_id='print_airflow_variable',
        python_callable=print_variable,
    )

 

1. Airflow UI 접속 → Admin > Variables 클릭
2. Key = filename, Value = input_file.txt 추가
3. DAG 실행 → 로그에서 filename이 출력됨

 

 

varialbe이 있는 경우 -> input_file.txt

Log message source details: sources=["/opt/airflow/logs/dag_id=variable_example/run_id=scheduled__2025-06-29T09:53:05.436680+00:00/task_id=print_airflow_variable/attempt=1.log"]
[2025-06-29, 18:53:15] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-06-29, 18:53:15] INFO - Filling up the DagBag from /opt/airflow/dags/variable_example.py: source="airflow.models.dagbag.DagBag"
[2025-06-29, 18:53:15] WARNING - /home/airflow/.local/lib/python3.12/site-packages/airflow/models/variable.py:147: DeprecationWarning: Using Variable.get from `airflow.models` is deprecated.Please use `from airflow.sdk import Variable` instead
  warnings.warn(
: source="py.warnings"
[2025-06-29, 18:53:15] INFO - Done. Returned value was: None: source="airflow.task.operators.airflow.providers.standard.operators.python.PythonOperator"
[2025-06-29, 18:53:15] INFO - 📂 Airflow Variable 'filename' is: input_file.txt: chan="stdout": source="task"

 

varialbe이 없는 경우 -> default인  default.txt

Log message source details: sources=["/opt/airflow/logs/dag_id=variable_example/run_id=manual__2025-06-29T09:53:54.881342+00:00/task_id=print_airflow_variable/attempt=1.log"]
[2025-06-29, 18:53:55] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-06-29, 18:53:55] INFO - Filling up the DagBag from /opt/airflow/dags/variable_example.py: source="airflow.models.dagbag.DagBag"
[2025-06-29, 18:53:55] WARNING - /home/airflow/.local/lib/python3.12/site-packages/airflow/models/variable.py:147: DeprecationWarning: Using Variable.get from `airflow.models` is deprecated.Please use `from airflow.sdk import Variable` instead
  warnings.warn(
: source="py.warnings"
[2025-06-29, 18:53:55] INFO - Done. Returned value was: None: source="airflow.task.operators.airflow.providers.standard.operators.python.PythonOperator"
[2025-06-29, 18:53:55] INFO - 📂 Airflow Variable 'filename' is: default.txt: chan="stdout": source="task"

 

메일보내기

 

먼저 메일 자체를 확인 해보자

import smtplib

smtp_host = 'smtp.gmail.com'
smtp_port = 587
username = 'name@gmail.com'       # 본인의 Gmail 주소
password = '16자리 비밀번호'    # Gmail 앱 비밀번호 

server = smtplib.SMTP(smtp_host, smtp_port)
server.starttls()
server.login(username, password)

from email.mime.text import MIMEText

msg = MIMEText('This is a test email from Python SMTP')
msg['Subject'] = 'Test Email'
msg['From'] = username
msg['To'] = 'name@naver.com'   # 받는 사람 이메일 주소

server.sendmail(username, ['name@naver.com'], msg.as_string())
server.quit()
print("Email sent successfully!")

메일 잘 온다.

 

이제 DAG로 진행해보자

from airflow import DAG
from airflow.operators.email import EmailOperator
from airflow.operators.empty import EmptyOperator
from datetime import datetime, timedelta

with DAG(
    dag_id='email_example',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(minutes=10),  
    catchup=False,
    tags=['email'],
) as dag:

    start = EmptyOperator(task_id='start')

    send_email = EmailOperator(
        task_id='send_email_task',
        to='name@naver.com',
        subject='Airflow Email Test',
        html_content="""
        <h3>📬 This is a test email sent from Airflow DAG</h3>
        <p>Hello! This email was sent automatically via EmailOperator.</p>
        """,
        from_email='name@gmail.com'
    )

    end = EmptyOperator(task_id='end')

    start >> send_email >> end

    # SMTP
    AIRFLOW__SMTP__SMTP_HOST: smtp.gmail.com
    AIRFLOW__SMTP__SMTP_PORT: 587
    AIRFLOW__SMTP__SMTP_STARTTLS: 'True'
    AIRFLOW__SMTP__SMTP_SSL: 'False'
    AIRFLOW__SMTP__SMTP_USER: name@gmail.com
    AIRFLOW__SMTP__SMTP_PASSWORD: <16자리 비밀번호>
    AIRFLOW__SMTP__SMTP_MAIL_FROM: name@gmail.com

도커 컴포즈 환경변수에 SMTP 설정을 넣었는데도 오류가 나는 이유는, Airflow EmailOperator가 실제로 메일을 보내기 위해서는 ‘Connection’이라는 별도의 설정이 Airflow 내부에 반드시 등록되어 있어야 한다. 

• 도커 컴포즈 환경변수에 SMTP 정보 넣으면 Airflow 서버 환경에는 반영되지만,
• Airflow 내부에서 conn_id='smtp_default'라는 Connection 설정이 따로 있어야 이메일 보내기가 가능하다.
• 이 Connection 설정은 Airflow UI(Admin > Connections) 에서 직접 등록하거나, CLI 또는 코드로 생성하면 된다.

 

왜 이게 필요한가?

• Airflow EmailOperator는 내부에서 smtp_default라는 Connection 정보를 찾는다.
• 그 Connection이 없으면 "Connection not found" 오류가 발생
• 도커 컴포즈에 넣는 환경변수는 SMTP 서버 기본 정보(호스트, 포트, 사용자 등)를 제공하는 역할
• Airflow UI의 Connections 메뉴에서 smtp_default를 등록하는 건 Airflow가 내부적으로 메일 서버 접속 정보를 관리하는 방법

 

Airflow 내부적으로 EmailOperator가 SMTP 서버 설정을 smtp_default라는  connection ID로  참조하기  때문

성공!!

 

Slack 연동

 

메일 보낼때 슬랙으로 메일보냈다고 알림 추가해보자

from airflow import DAG
from airflow.operators.email import EmailOperator
from airflow.operators.empty import EmptyOperator
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
from datetime import datetime, timedelta

with DAG(
    dag_id='email_with_slack_example',
    start_date=datetime(2025, 6, 29),
    schedule=timedelta(minutes=10),
    catchup=False,
    tags=['email', 'slack'],
) as dag:

    start = EmptyOperator(task_id='start')

    send_email = EmailOperator(
        task_id='send_email_task',
        to='name@naver.com',
        subject='Airflow Email Test',
        html_content="""
        <h3>📬 This is a test email sent from Airflow DAG</h3>
        <p>Hello! This email was sent automatically via EmailOperator.</p>
        """,
        from_email='name@gmail.com'
    )

    notify_slack = SlackWebhookOperator(
        task_id='notify_slack_task',
        slack_webhook_conn_id="slack_conn",  
        message="✅ Email successfully sent to name@gmail.com from Airflow!",
        username='airflow',
        channel="#airflow_practice",  

    )

    end = EmptyOperator(task_id='end')

    start >> send_email >> notify_slack >> end

docker compose run --rm airflow-cli connections add slack_conn \
  --conn-type slackwebhook \
  --conn-password 'https://hooks.slack.com/services/XXX/YYY/ZZZ'

 

Slack Webhook의 URL(=endpoint)을 Airflow 컨테이너 안에 Connection으로 등록하여 DAG에서 슬랙으로 알림을 보낼 수 있도록 하는 것

성공적으로 메일이 온다.

 

jinja

 

참고: DAG정의 내용이 길어질때

# default_args 없이 with 안에 다 넣는 방식
with DAG(
    'jinja_template_example',
    owner='airflow',
    start_date=datetime(2025, 6, 30),
    retries=1,
    retry_delay=timedelta(minutes=5),
    schedule='@daily',
    catchup=False,
    tags=['example'],
) as dag:

# default_args 따로 빼는 방식
default_args = {
    'owner': 'airflow',
    'start_date': datetime(2025, 6, 30),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'jinja_template_example',
    default_args=default_args,
    schedule='@daily',
    catchup=False,
    tags=['example'],
) as dag:

 

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


default_args = {
    'owner': 'airflow',
    'start_date': datetime(2025, 6, 30),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'jinja_template_example',
    default_args=default_args,
    schedule='@daily',
    catchup=False,
    tags=['example'],
) as dag:

    # BashOperator에서 ds 변수를 이용해 출력하기
    templated_command = """
    echo "오늘 날짜는 {{ ds }} 입니다."
    echo "7일 후 날짜는 {{ macros.ds_add(ds, 7) }}"
    echo "파라미터로 받은 값: {{ params.my_param }}"
    """

    run_bash = BashOperator(
        task_id='print_dates',
        bash_command=templated_command,
        params={'my_param': 'Hello Airflow!'}
    )

 

• {{ ds }} : Airflow 내장 변수로 DAG 실행 날짜(yyyy-mm-dd) 가 들어감
• {{ macros.ds_add(ds, 7) }} : ds 날짜에 7일을 더한 날짜를 출력. macros는 Airflow가 제공하는 여러 유틸 함수 모음
• {{ params.my_param }} : params 딕셔너리에 넘긴 사용자 지정 파라미터를 출력

[2025-06-30, 11:16:24] INFO - Running command: ['/usr/bin/bash', '-c', '\n    echo "오늘 날짜는 2025-06-30 입니다."\n    echo "7일 후 날짜는 2025-07-07"\n    echo "파라미터로 받은 값: Hello Airflow!"\n    ']: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"

 

• Airflow에서 Jinja 템플릿이란?
• Airflow는 Python으로 DAG를 정의한다. -> DAG 안에서 파라미터, 명령어, SQL 등을 문자열로 작성
• 이 문자열을 실행 시점에 날짜나 변수에 맞게 바꾸는(동적 처리) 기능이 필요
• 그래서 Airflow는 Python에서 널리 쓰이는 템플릿 엔진인 Jinja2를 사용