1 asked

1

압축파일을 윈도우기본 압축 프로그램으로 풀었더니 실습코드 폴더가 안나오는 현상이 있었습니다! 실습코드가 없어진줄 알있는데 실습파일 다시 잘 확인했습니다~

0

슬랙 실습

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
import json


def send_error_to_slack(context):
    """
    에러 발생 시 Slack으로 메시지 전송
    """

    # 에러 정보 추출
    task_instance = context.get('task_instance')
    dag_id = context.get('dag').dag_id
    task_id = task_instance.task_id
    execution_date = context.get('execution_date')
    exception = context.get('exception')

    # Slack 메시지 구성
    slack_message = {
        "text": "❌ Airflow Task Failed",
        "attachments": [
            {
                "color": "danger",
                "fields": [
                    {
                        "title": "DAG",
                        "value": dag_id,
                        "short": True
                    },
                    {
                        "title": "Task",
                        "value": task_id,
                        "short": True
                    },
                    {
                        "title": "Time",
                        "value": str(execution_date),
                        "short": True
                    },
                    {
                        "title": "Error Message",
                        "value": str(exception) if exception else "Unknown error",
                        "short": False
                    }
                ]
            }
        ]
    }

    # Slack으로 전송
    try:
        slack_hook = SlackWebhookHook(
            slack_webhook_conn_id='slack_default'
        )
        slack_hook.send_dict(slack_message)
        print("Error notification sent to Slack")
    except Exception as e:
        print(f"Failed to send Slack notification: {e}")


# 테스트용 함수들
def task_that_fails():
    """실패하는 태스크"""
    raise Exception("This task intentionally fails for testing")


def task_that_succeeds():
    """성공하는 태스크 (알림 없음)"""
    print("Task completed successfully")
    return "success"


# DAG 설정
default_args = {
    'owner': 'data-team',
    'start_date': datetime(2024, 1, 1),
    'retries': 0,  # 재시도 없음
    'on_failure_callback': send_error_to_slack,  # 에러 시에만 알림
}

dag = DAG(
    'slack_error_notification',
    default_args=default_args,
    description='에러 발생 시 Slack 알림',
    schedule_interval='@daily',
    catchup=False
)

# 태스크 정의
fail_task = PythonOperator(
    task_id='failing_task',
    python_callable=task_that_fails,
    dag=dag
)

success_task = PythonOperator(
    task_id='success_task',
    python_callable=task_that_succeeds,
    dag=dag
)

# 태스크 실행 순서
success_task >> fail_task

0

혹시 dellahong님 어떤 root.py가 보이지 않는다고 하시는지 공유 가능하실까요??

0

Parallel

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
import logging
import time
import random
from concurrent.futures import ThreadPoolExecutor, as_completed

default_args = {
    'owner': 'data-team',
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=2),
}

def generate_data(total_records=500):
    symbols = ['AAPL', 'GOOGL', 'MSFT']
    data = []
    for i in range(total_records):
        data.append({
            'id': i + 1,
            'symbol': random.choice(symbols),
            'price': round(random.uniform(100, 300), 2),
            'volume': random.randint(1000, 5000)
        })
    logging.info(f"데이터 생성: {len(data)}건")
    return data


def create_batches(data, batch_size):
    batches = []
    for i in range(0, len(data), batch_size):
        batches.append(data[i:i + batch_size])
    logging.info(f"배치 생성: {len(batches)}개")
    return batches

def process_batch(batch_data, batch_id):
    time.sleep(1) #일련의 데이터 작업이 보통 들어감 
    total_volume = sum(record['volume'] for record in batch_data)
    result = {
        'batch_id': batch_id,
        'count': len(batch_data),
        'total_volume': total_volume
    }
    logging.info(f"배치 {batch_id} 완료")
    return result

def parallel_processing():
    start_time = time.time()
    data = generate_data(500)
    batches = create_batches(data, 50)

    results = []
    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = {
            executor.submit(process_batch, batch, f"batch_{i+1}"): i
            for i, batch in enumerate(batches)
        }

    for future in as_completed(futures):
        result = future.result()
        results.append(result)

    processing_time = round(time.time() - start_time, 2)
    logging.info(f"병렬 처리 완료: {processing_time}초, 총 {len(results)}개 배치")
    return {
        'time': processing_time,
        'batches': len(batches),
        'results': results
    }

parallel_dag = DAG(
    'parallel_processing_dag',
    default_args=default_args,
    description='병렬 처리 예제',
    schedule='@daily',
    catchup=False,
    tags=['parallel']
)

start = DummyOperator(task_id='start', dag=parallel_dag)

parallel_task = PythonOperator(
    task_id='parallel_task',
    python_callable=parallel_processing,
    dag=parallel_dag
)

end = DummyOperator(task_id='end', dag=parallel_dag)

start >> parallel_task >> end

0

Custom Operator

from datetime import datetime, timedelta
from airflow import DAG
from airflow.models import BaseOperator
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.decorators import apply_defaults
from db_utils import execute_query
import logging

#기본 설정
default_args = {
    'owner': 'data-team',
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=2),
}

custom_operator_dag = DAG(
    'custom_operator_dag',
    default_args=default_args,
    description='Custom Operator 간단예제',
    schedule='@daily',
    catchup=False,
    tags=['custom-operator', 'simple']
)

start = DummyOperator(task_id='start', dag=custom_operator_dag)

# ================= Custom Operator =================

class StockDataOperator(BaseOperator):
    """주식 데이터를 조회하는 커스텀 Operator"""

    @apply_defaults
    def __init__(self, symbol, days=7, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.symbol = symbol
        self.days = days

    def execute(self, context):
        """실제 실행되는 메소드"""
        logging.info(f"{self.symbol} 주식 데이터 조회 시작")

        query = """
        SELECT symbol, COUNT(*) as count, AVG(price) as avg_price
        FROM stock_prices
        WHERE symbol = %s
        AND trade_date >= CURDATE() - INTERVAL %s DAY
        GROUP BY symbol
        """

        result = execute_query(query, [self.symbol, self.days])

        if result:
            data = {
                'symbol': result[0][0],
                'count': result[0][1],
                'avg_price': float(result[0][2]) if result[0][2] else 0.0
            }
            logging.info(f"{self.symbol} 조회완료 {data}")
            return data
        else:
            logging.warning(f"{self.symbol} 데이터가 없습니다")
            return {'symbol': self.symbol, 'count': 0, 'avg_price': 0.0}

#Custom Operator 사용
custom_task = StockDataOperator(
    task_id='extract_stock_data',
    symbol='APPL',
    days=30,
    dag=custom_operator_dag
)

# ================= 기존 Python Operator =================
 
 
# 기존 PythonOperator와 비교
def traditional_function(**context):
    """기존 방식의 함수"""
    logging.info("기존 PythonOperator 방식")

    query = "SELECT COUNT(*) FROM stock_prices WHERE symbol = 'AAPL'"
    result = execute_query(query)
    count = result[0][0] if result else 0

    return {'symbol': 'AAPL', 'count': count}

traditional_task = PythonOperator(
    task_id='traditional_approach',
    python_callable=traditional_function,
    dag=custom_operator_dag
)

end = DummyOperator(task_id='end', dag=custom_operator_dag)

#Task 의존성
start >> [custom_task, traditional_task] >> end
    

0

TaskGroup

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.task_group import TaskGroup
from db_utils import execute_query
import logging

#기본 설정
default_args = {
    'owner': 'data-team',
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=2),
}

def extract_data(symbol, **context):
    """데이터 추출"""
    logging.info(f"{symbol} 데이터 추출")

    query = "SELECT COUNT(*) FROM stock_prices WHERE symbol = %s"
    result = execute_query(query, [symbol])

    count = result[0][0] if result else 0
    logging.info(f"{symbol}: {count}건 추출")
    return {'symbol': symbol, 'count': count}


def process_data(symbol, **context):
    """데이터 처리"""
    logging.info(f"{symbol} 데이터 처리")

    ti = context['ti']  # Task Instance (TI)
    extracted = ti.xcom_pull(task_ids=f'extract_group.extract_{symbol.lower()}')

    processed = {
        'symbol': symbol,
        'count': extracted['count'],
        'status': 'processed' if extracted['count'] > 0 else 'empty'
    }

    logging.info(f"{symbol} 처리 완료: {processed}")
    return processed

def generate_report(**context):
    """전체 리포트 생성"""
    logging.info("리포트 생성 시작")

    ti = context['ti']
    symbols = ['AAPL', 'GOOGL']

    total_count = 0
    for symbol in symbols:
        processed = ti.xcom_pull(task_ids=f'process_group.nested_group.process_{symbol.lower()}')
        if processed:
            total_count += processed['count']

    report = f"총 {total_count}건 데이터 처리 완료"
    logging.info(report)
    return report

# TaskGroup DAG
taskgroup_dag = DAG(
    'taskgroup_simple_dag',
    default_args=default_args,
    description='TaskGroup 간단 예제',
    schedule='@daily',
    catchup=False,
    tags=['taskgroup', 'simple']
)

start = DummyOperator(task_id='start', dag=taskgroup_dag)

# 기본 Task Group
with TaskGroup(group_id='extract_group', dag=taskgroup_dag) as extract_group:

    symbols = ['AAPL', 'GOOGL']

    for symbol in symbols:
        PythonOperator(
            task_id=f'extract_{symbol.lower()}',
            python_callable=extract_data,
            op_kwargs={'symbol': symbol},
            dag=taskgroup_dag
        )

with TaskGroup(group_id='process_group', dag=taskgroup_dag) as process_group:

    # 중첩된 TaskGroup
    with TaskGroup(group_id='nested_group', dag=taskgroup_dag) as nested_group:

        for symbol in symbols:
            PythonOperator(
                task_id=f'process_{symbol.lower()}',
                python_callable=process_data,
                op_kwargs={'symbol': symbol},
                dag=taskgroup_dag
            )

    # 리포트 태스크
    report_task = PythonOperator(
        task_id='generate_report',
        python_callable=generate_report,
        dag=taskgroup_dag
    )

    # 중첩 그룹 완료 후 리포트 실행
    nested_group >> report_task

end = DummyOperator(task_id='end', dag=taskgroup_dag)

start >> extract_group >> process_group >> end

0

잠시 확인하고 말씀드리겠습니다. 감사합니다!