BashOperator 예시:

from airflow.operators.bash_operator import BashOperator

create_folder = BashOperator(
    task_id='create_folder',
    bash_command='mkdir -p /tmp/data/{{ ds }}',
    dag=dag
)

PythonOperator 예시:

from airflow.operators.python_operator import PythonOperator

def process_data(ds, **kwargs):
    print(f"데이터 처리 중... 날짜: {ds}")
    return "처리 완료!"

process_task = PythonOperator(
    task_id='process_data',
    python_callable=process_data,
    dag=dag
)

hello_world_dag.py:

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

# DAG의 기본 인자 설정
default_args = {
    'owner': '여러분의_이름',
    'depends_on_past': False,
    'email': ['your_email@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG 정의
dag = DAG(
    'hello_world',
    default_args=default_args,
    description='첫 번째 DAG',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 1, 1),
    catchup=False,
    tags=['example'],
)

# Task 1: Hello 출력
task_hello = BashOperator(
    task_id='print_hello',
    bash_command='echo "Hello"',
    dag=dag,
)

# Task 2: 잠시 기다리기
def wait_function():
    import time
    time.sleep(5)
    return "기다림 완료!"

task_wait = PythonOperator(
    task_id='wait_5_seconds',
    python_callable=wait_function,
    dag=dag,
)

# Task 3: World 출력
task_world = BashOperator(
    task_id='print_world',
    bash_command='echo "World"',
    dag=dag,
)

# 작업 순서 정의
task_hello >> task_wait >> task_world

weather_data_dag.py:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import requests
import json
import pandas as pd

default_args = {
    'owner': '여러분의_이름',
    'depends_on_past': False,
    'email': ['your_email@example.com'],
    'email_on_failure': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'weather_data_pipeline',
    default_args=default_args,
    description='날씨 데이터 수집 및 처리',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 1, 1),
    catchup=False,
    tags=['weather', 'data'],
)

# 날씨 데이터 수집 함수
def fetch_weather_data(**context):
    # 서울의 날씨 데이터 가져오기 (OpenWeatherMap API 사용)
    api_key = "여기에_API_키_입력"  # 실제 사용 시 API 키 필요
    city = "Seoul"
    url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"
    
    # API 호출 (실제 API 키가 없으므로 예시 데이터 사용)
    # response = requests.get(url)
    # data = response.json()
    
    # 예시 데이터
    data = {
        "main": {
            "temp": 25.05,
            "feels_like": 24.8,
            "temp_min": 23.33,
            "temp_max": 26.11,
            "pressure": 1015,
            "humidity": 57
        },
        "wind": {
            "speed": 2.57,
            "deg": 180
        },
        "clouds": {
            "all": 75
        },
        "dt": 1629343200,
        "sys": {
            "sunrise": 1629317125,
            "sunset": 1629366167
        },
        "name": "Seoul"
    }
    
    # 결과를 XCom에 저장 (다음 작업에서 사용)
    context['ti'].xcom_push(key='weather_data', value=data)
    return "날씨 데이터 수집 완료!"

# 데이터 처리 함수
def process_weather_data(**context):
    # 이전 작업에서 저장한 데이터 가져오기
    ti = context['ti']
    data = ti.xcom_pull(task_ids='fetch_weather', key='weather_data')
    
    # 필요한 정보 추출
    processed_data = {
        "city": data["name"],
        "temperature": data["main"]["temp"],
        "humidity": data["main"]["humidity"],
        "wind_speed": data["wind"]["speed"],
        "date": datetime.fromtimestamp(data["dt"]).strftime('%Y-%m-%d')
    }
    
    # 결과를 XCom에 저장
    ti.xcom_push(key='processed_data', value=processed_data)
    return "데이터 처리 완료!"

# 결과 저장 함수
def save_results(**context):
    # 처리된 데이터 가져오기
    ti = context['ti']
    data = ti.xcom_pull(task_ids='process_data', key='processed_data')
    
    # CSV 파일로 저장 (실제로는 저장하지 않고 출력만)
    print(f"데이터를 저장합니다: {data}")
    
    # 판다스 DataFrame으로 변환 (실제로는 파일로 저장)
    # df = pd.DataFrame([data])
    # df.to_csv(f"/tmp/weather_{data['date']}.csv", index=False)
    
    return "결과 저장 완료!"

# 작업 정의
fetch_weather = PythonOperator(
    task_id='fetch_weather',
    python_callable=fetch_weather_data,
    dag=dag,
)

process_data = PythonOperator(
    task_id='process_data',
    python_callable=process_weather_data,
    dag=dag,
)

save_results = PythonOperator(
    task_id='save_results',
    python_callable=save_results,
    dag=dag,
)

# 작업 순서 정의
fetch_weather >> process_data >> save_results

다양한 스케줄링 예시:

# 매일 오전 8시에 실행
dag = DAG('daily_morning', schedule_interval='0 8 * * *', ...)

# 매주 월요일 자정에 실행
dag = DAG('weekly_report', schedule_interval='0 0 * * 1', ...)

# 매월 1일 오후 3시에 실행
dag = DAG('monthly_summary', schedule_interval='0 15 1 * *', ...)

# 매시간 실행
dag = DAG('hourly_check', schedule_interval='@hourly', ...)

# 5분마다 실행
dag = DAG('frequent_check', schedule_interval=timedelta(minutes=5), ...)

복잡한 의존성 예시:

# 팬아웃 (1 → 다수)
task_start >> [task_a, task_b, task_c]

# 팬인 (다수 → 1)
[task_a, task_b, task_c] >> task_end

# 브랜치 (조건부 실행)
from airflow.operators.python import BranchPythonOperator

def choose_path(**context):
    if context['execution_date'].day % 2 == 0:
        return 'even_day_task'
    else:
        return 'odd_day_task'

branch_task = BranchPythonOperator(
    task_id='branch_task',
    python_callable=choose_path,
    dag=dag,
)

even_day_task = BashOperator(...)
odd_day_task = BashOperator(...)

branch_task >> [even_day_task, odd_day_task]

# 트리거 규칙
from airflow.utils.trigger_rule import TriggerRule

cleanup_task = BashOperator(
    task_id='cleanup',
    bash_command='echo "Cleaning up..."',
    trigger_rule=TriggerRule.ALL_DONE,  # 모든 상위 작업이 완료되면 실행 (성공/실패 무관)
    dag=dag,
)

실패 처리 예시:

# 재시도 설정
task_with_retry = BashOperator(
    task_id='task_with_retry',
    bash_command='exit 1',  # 일부러 실패하는 명령어
    retries=3,  # 최대 3번 재시도
    retry_delay=timedelta(minutes=5),  # 5분 간격으로 재시도
    dag=dag,
)

# 타임아웃 설정
task_with_timeout = BashOperator(
    task_id='task_with_timeout',
    bash_command='sleep 300',  # 5분 동안 실행
    execution_timeout=timedelta(minutes=2),  # 2분 후 타임아웃
    dag=dag,
)

# 이메일 알림 설정
dag = DAG(
    'email_notification_dag',
    default_args={
        'email': ['your_email@example.com'],
        'email_on_failure': True,
        'email_on_retry': False,
    },
    ...
)

# 콜백 함수
def on_success_callback(context):
    print(f"Task {context['task_instance'].task_id} succeeded!")

def on_failure_callback(context):
    print(f"Task {context['task_instance'].task_id} failed!")
    # Slack, Teams 등으로 알림을 보낼 수도 있음

task_with_callback = BashOperator(
    task_id='task_with_callback',
    bash_command='echo "Hello"',
    on_success_callback=on_success_callback,
    on_failure_callback=on_failure_callback,
    dag=dag,
)

# SLA 설정
task_with_sla = BashOperator(
    task_id='task_with_sla',
    bash_command='sleep 60',
    sla=timedelta(seconds=30),  # 30초 내에 완료되어야 함
    dag=dag,
)

최적화된 센서 사용 예시:

from airflow.sensors.external_task import ExternalTaskSensor

# 비효율적인 방법 (기본 mode='poke')
wait_for_task = ExternalTaskSensor(
    task_id='wait_for_other_dag',
    external_dag_id='other_dag',
    external_task_id='final_task',
    dag=dag,
)

# 최적화된 방법 (mode='reschedule')
wait_for_task_optimized = ExternalTaskSensor(
    task_id='wait_for_other_dag_optimized',
    external_dag_id='other_dag',
    external_task_id='final_task',
    mode='reschedule',  # 작업 슬롯을 차지하지 않고 대기
    dag=dag,
)

Prometheus 메트릭 설정 예시:

# airflow.cfg 파일에 추가
[metrics]
statsd_on = True
statsd_host = localhost
statsd_port = 8125
statsd_prefix = airflow

DAG 팩토리 패턴 예시:

def create_data_pipeline_dag(
    dag_id,
    schedule_interval,
    source_system,
    target_system,
    start_date=datetime(2023, 1, 1)
):
    """여러 데이터 파이프라인 DAG를 생성하는 팩토리 함수"""
    
    dag = DAG(
        dag_id=dag_id,
        schedule_interval=schedule_interval,
        start_date=start_date,
        catchup=False,
        tags=[source_system, target_system, 'data_pipeline'],
    )
    
    with dag:
        extract = PythonOperator(
            task_id=f'extract_from_{source_system}',
            python_callable=extract_data,
            op_kwargs={'source': source_system},
        )
        
        transform = PythonOperator(
            task_id='transform_data',
            python_callable=transform_data,
        )
        
        load = PythonOperator(
            task_id=f'load_to_{target_system}',
            python_callable=load_data,
            op_kwargs={'target': target_system},
        )
        
        extract >> transform >> load
    
    return dag

# 여러 DAG 생성
sales_dag = create_data_pipeline_dag(
    dag_id='sales_data_pipeline',
    schedule_interval='@daily',
    source_system='sales_db',
    target_system='data_warehouse'
)

marketing_dag = create_data_pipeline_dag(
    dag_id='marketing_data_pipeline',
    schedule_interval='@daily',
    source_system='marketing_api',
    target_system='data_warehouse'
)