데이터 과학 워크플로우: Airflow를 이용한 파이프라인 자동화 🚀

안녕하세요, 데이터 과학 열정가 여러분! 오늘은 정말 흥미진진한 주제로 여러분과 함께 데이터의 세계를 탐험해보려고 해요. 바로 Airflow를 이용한 데이터 파이프라인 자동화에 대해 알아볼 거예요. 🎉

여러분, 혹시 데이터 처리 작업을 하다가 "아, 이 과정을 자동화할 수 있다면 얼마나 좋을까?"라고 생각해본 적 있나요? 그렇다면 여러분은 이미 Airflow의 필요성을 느끼고 계신 거예요! Airflow는 마치 우리의 데이터 작업을 위한 슈퍼 영웅과 같아요. 복잡한 데이터 처리 과정을 자동화하고, 효율적으로 관리할 수 있게 해주는 강력한 도구랍니다. 😎

이 글을 통해 우리는 Airflow의 세계로 깊숙이 들어가 볼 거예요. 마치 재능넷에서 다양한 재능을 탐험하듯이, 우리도 Airflow의 다양한 기능과 활용법을 탐험해볼 거예요. 여러분의 데이터 과학 실력을 한 단계 업그레이드할 준비 되셨나요? 그럼 시작해볼까요! 🚀

1. Airflow란 무엇인가? 🤔

자, 여러분! Airflow에 대해 들어보셨나요? 아직 모르신다고요? 걱정 마세요. 지금부터 차근차근 설명해드릴게요. 🙂

Airflow는 워크플로우 관리 플랫폼이에요. 복잡하게 들리시나요? 쉽게 말해서, Airflow는 우리가 해야 할 일들을 순서대로 정리하고 자동으로 실행해주는 똑똑한 비서 같은 존재예요. 마치 재능넷에서 여러분이 원하는 재능을 찾아 순서대로 배우는 것처럼, Airflow는 데이터 처리 작업을 순서대로 실행해줍니다.

Airflow는 2014년 Airbnb에서 시작된 프로젝트로, 현재는 Apache Software Foundation의 최상위 프로젝트로 성장했어요. 마치 재능넷이 다양한 재능을 공유하는 플랫폼으로 성장한 것처럼, Airflow도 데이터 엔지니어링 분야에서 없어서는 안 될 중요한 도구로 자리잡았답니다. 👏

그럼 Airflow가 어떻게 작동하는지 좀 더 자세히 알아볼까요?

위의 그림에서 볼 수 있듯이, Airflow는 데이터 수집부터 결과 저장까지의 전체 과정을 관리해요. 마치 재능넷에서 여러분이 원하는 재능을 찾아 학습하고, 그 결과를 포트폴리오로 만드는 과정과 비슷하답니다. 😊

Airflow를 사용하면 다음과 같은 이점을 얻을 수 있어요:

• 자동화: 반복적인 작업을 자동으로 처리할 수 있어요.
• 스케줄링: 작업을 원하는 시간에 실행할 수 있어요.
• 의존성 관리: 작업 간의 순서와 의존 관계를 쉽게 정의할 수 있어요.
• 모니터링: 작업의 진행 상황을 실시간으로 확인할 수 있어요.

이제 Airflow가 무엇인지 대략적으로 이해하셨나요? 그렇다면 이제 Airflow의 핵심 개념들을 하나씩 살펴보도록 해요. 마치 재능넷에서 새로운 재능을 배우는 것처럼, 우리도 Airflow의 새로운 개념들을 하나씩 배워볼까요? 😃

2. Airflow의 핵심 개념 🔑

Airflow를 제대로 이해하기 위해서는 몇 가지 핵심 개념을 알아야 해요. 마치 재능넷에서 새로운 재능을 배우기 위해 기본 개념부터 시작하는 것처럼요. 자, 그럼 하나씩 살펴볼까요? 🧐

2.1 DAG (Directed Acyclic Graph) 📊

DAG는 Airflow의 가장 중요한 개념이에요. DAG는 '방향성 비순환 그래프'라는 뜻인데, 쉽게 말해 작업들의 실행 순서를 정의하는 청사진이라고 생각하면 돼요.

DAG를 이해하기 위해 간단한 예를 들어볼게요. 여러분이 재능넷에서 새로운 재능을 배우는 과정을 생각해보세요:

1. 재능 검색
2. 재능 선택
3. 수업 신청
4. 수업 수강
5. 과제 제출
6. 피드백 받기

이 과정을 DAG로 표현하면 다음과 같아요:

이 DAG에서 각 원은 하나의 Task(작업)를 나타내고, 화살표는 작업 간의 의존성을 나타내요. 이렇게 DAG를 사용하면 복잡한 워크플로우를 쉽게 표현하고 관리할 수 있답니다. 😊

2.2 Operator 🎭

Operator는 DAG 안에서 실제로 실행되는 작업의 단위예요. 각 Operator는 특정 작업을 수행하도록 설계되어 있죠. Airflow에는 다양한 종류의 Operator가 있어요:

• BashOperator: Bash 명령어를 실행
• PythonOperator: Python 함수를 실행
• SQLOperator: SQL 쿼리를 실행
• EmailOperator: 이메일을 보냄
• SimpleHttpOperator: HTTP 요청을 보냄
• 그 외 다양한 커스텀 Operator

예를 들어, 재능넷에서 새로운 재능을 검색하는 작업을 PythonOperator로 구현한다면 다음과 같이 할 수 있어요:

from airflow.operators.python_operator import PythonOperator

def search_talent():
    # 재능 검색 로직
    print("재능을 검색합니다.")

search_talent_task = PythonOperator(
    task_id='search_talent',
    python_callable=search_talent,
    dag=dag
)

이렇게 Operator를 사용하면 다양한 종류의 작업을 쉽게 정의하고 실행할 수 있어요. 마치 재능넷에서 다양한 재능을 쉽게 찾고 배울 수 있는 것처럼 말이죠! 😉

2.3 Task 📌

Task는 DAG 안에서 실행되는 개별 작업의 인스턴스를 말해요. 쉽게 말해, Operator를 사용해 정의한 작업이 실제로 DAG 안에서 실행될 때, 그것을 Task라고 부른답니다.

Task들은 서로 의존성을 가질 수 있어요. 예를 들어, '재능 검색' Task가 완료된 후에 '재능 선택' Task가 실행되어야 한다면, 이런 의존성을 다음과 같이 표현할 수 있어요:

search_talent_task >> select_talent_task

이 표현은 "search_talent_task가 select_talent_task보다 먼저 실행되어야 한다"는 의미예요. 이렇게 Task 간의 의존성을 정의하면, Airflow가 자동으로 올바른 순서로 Task를 실행해줍니다. 👍

2.4 Executor 🏃‍♂️

Executor는 Task를 어떻게 실행할지 결정하는 구성 요소예요. Airflow에는 여러 종류의 Executor가 있어요:

• SequentialExecutor: 기본 Executor, Task를 순차적으로 실행
• LocalExecutor: 로컬 머신에서 병렬로 Task 실행
• CeleryExecutor: 여러 워커 노드에서 분산 처리
• KubernetesExecutor: Kubernetes 클러스터에서 Task 실행

Executor를 선택할 때는 여러분의 워크로드와 인프라 환경을 고려해야 해요. 마치 재능넷에서 여러분의 학습 스타일에 맞는 강사를 선택하는 것처럼 말이죠! 🤓

2.5 XCom 💌

XCom(Cross-Communication)은 Task 간에 작은 양의 데이터를 주고받을 수 있게 해주는 기능이에요. 예를 들어, '재능 검색' Task에서 찾은 재능의 ID를 '재능 선택' Task로 전달하고 싶다면 XCom을 사용할 수 있어요.

def search_talent(**kwargs):
    talent_id = 12345  # 예시 ID
    kwargs['ti'].xcom_push(key='talent_id', value=talent_id)

def select_talent(**kwargs):
    talent_id = kwargs['ti'].xcom_pull(key='talent_id', task_ids='search_talent')
    print(f"선택된 재능 ID: {talent_id}")

search_task = PythonOperator(
    task_id='search_talent',
    python_callable=search_talent,
    provide_context=True,
    dag=dag
)

select_task = PythonOperator(
    task_id='select_talent',
    python_callable=select_talent,
    provide_context=True,
    dag=dag
)

search_task >> select_task

이렇게 XCom을 사용하면 Task 간에 필요한 정보를 쉽게 공유할 수 있어요. 마치 재능넷에서 강사와 학생이 정보를 주고받는 것처럼요! 😊

지금까지 Airflow의 핵심 개념들을 살펴봤어요. 이 개념들을 잘 이해하면 Airflow를 사용해 복잡한 데이터 파이프라인도 쉽게 구축할 수 있답니다. 다음 섹션에서는 이런 개념들을 바탕으로 실제 Airflow를 설치하고 사용하는 방법에 대해 알아볼 거예요. 준비되셨나요? 그럼 계속해서 Airflow의 세계로 더 깊이 들어가 볼까요? 🚀

3. Airflow 설치 및 설정 🛠️

자, 이제 Airflow의 기본 개념을 이해했으니 실제로 Airflow를 설치하고 설정하는 방법을 알아볼까요? 마치 재능넷에서 새로운 재능을 배우기 위해 필요한 도구를 준비하는 것처럼, 우리도 Airflow를 사용하기 위한 준비를 해볼 거예요. 😊

3.1 Airflow 설치하기 📥

Airflow를 설치하는 방법은 여러 가지가 있지만, 가장 간단한 방법은 pip를 사용하는 거예요. 먼저 가상 환경을 만들고 그 안에 Airflow를 설치하는 것이 좋아요.

# 가상 환경 생성
python -m venv airflow_env

# 가상 환경 활성화 (Windows)
airflow_env\Scripts\activate

# 가상 환경 활성화 (Mac/Linux)
source airflow_env/bin/activate

# Airflow 설치
pip install apache-airflow

주의: Airflow 설치 과정에서 많은 의존성 패키지들이 함께 설치되므로 시간이 좀 걸릴 수 있어요. 마치 재능넷에서 새로운 재능을 배우기 위해 여러 가지 준비물을 모으는 것처럼 말이죠! 😉

3.2 Airflow 초기화하기 🚀

Airflow를 설치했다면 이제 초기화를 해야 해요. 이 과정에서 Airflow에 필요한 기본 디렉토리와 설정 파일들이 생성됩니다.

# Airflow 홈 디렉토리 설정 (Windows)
set AIRFLOW_HOME=C:\Users\YourUsername\airflow

# Airflow 홈 디렉토리 설정 (Mac/Linux)
export AIRFLOW_HOME=~/airflow

# Airflow 데이터베이스 초기화
airflow db init

이 명령어를 실행하면 AIRFLOW_HOME 디렉토리에 Airflow 관련 파일들이 생성돼요. 주요 파일과 디렉토리는 다음과 같아요:

• airflow.cfg: Airflow 설정 파일
• airflow.db: 기본 SQLite 데이터베이스
• logs: 로그 파일 디렉토리
• dags: DAG 파일을 저장할 디렉토리

3.3 Airflow 웹 서버와 스케줄러 실행하기 🌐

Airflow를 사용하려면 웹 서버와 스케줄러를 실행해야 해요. 웹 서버는 Airflow의 UI를 제공하고, 스케줄러는 정의된 DAG를 실행하는 역할을 해요.

# 웹 서버 실행 (백그라운드에서 실행)
airflow webserver --port 8080 -D

# 스케줄러 실행 (백그라운드에서 실행)
airflow scheduler -D

이제 브라우저에서 http://localhost:8080으로 접속하면 Airflow UI를 볼 수 있어요. 마치 재능넷 웹사이트에 접속하는 것처럼 쉽죠? 😊

3.4 Airflow 설정 커스터마이징 🛠️

Airflow의 동작을 커스터마이징하고 싶다면 airflow.cfg 파일을 수정하면 돼요. 이 파일에서 다양한 설정을 변경할 수 있어요:

• executor: Task 실행 방식 설정
• sql_alchemy_conn: 데이터베이스 연결 설정
• dag_folder: DAG 파일 위치 설정
• load_examples: 예제 DAG 로드 여부 설정

예를 들어, 예제 DAG를 로드하지 않으려면 다음과 같이 설정을 변경할 수 있어요:

load_examples = False

주의: airflow.cfg 파일을 수정한 후에는 Airflow 서비스를 재시작해야 변경사항이 적용돼요.

3.5 Airflow CLI 사용하기 💻

Airflow는 강력한 명령줄 인터페이스(CLI)를 제공해요. CLI를 통해 다양한 Airflow 관련 작업을 수행할 수 있죠. 주요 명령어들을 살펴볼까요?

# DAG 목록 보기
airflow dags list

# 특정 DAG의 Task 목록 보기
airflow tasks list [dag_id]

# DAG 실행하기
airflow dags trigger [dag_id]

# DAG 일시 중지/재개하기
airflow dags pause [dag_id]
airflow dags unpause [dag_id]

# 특정 Task 상태 확인하기
airflow tasks state [dag_id] [task_id] [execution_date]

이런 CLI 명령어들을 사용하면 Airflow를 더욱 효과적으로 관리할 수 있어요. 마치 재능넷에서 다양한 기능을 사용해 학습을 관리하는 것처럼 말이죠! 😉

위 그림은 Airflow의 주요 구성 요소들을 보여줘요. 각 구성 요소들이 어떻게 상호작용하는지 이해하면 Airflow를 더 효과적으로 사용할 수 있답니다. 😊

이제 Airflow를 설치하고 기본적인 설정을 마쳤어요. 다음 섹션에서는 실제로 DAG를 작성하고 실행하는 방법에 대해 알아볼 거예요. 재능넷에서 새로운 재능을 배우기 시작하는 것처럼, 우리도 이제 Airflow로 데이터 파이프라인을 만들어볼 준비가 된 거예요! 🚀

4. 첫 번째 DAG 만들기 🎨

자, 이제 Airflow의 기본 설정을 마쳤으니 실제로 DAG를 만들어볼 차례예요. 마치 재능넷에서 첫 수업을 듣는 것처럼 설렘 가득한 마음으로 시작해볼까요? 😊

4.1 DAG 파일 생성하기 📝

먼저, AIRFLOW_HOME/dags 디렉토리에 새로운 Python 파일을 만들어요. 이 파일 이름을 my_first_dag.py라고 지어볼까요?

# AIRFLOW_HOME/dags/my_first_dag.py

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

# DAG의 기본 인자 설정
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email': ['your_email@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG 정의
dag = DAG(
    'my_first_dag',
    default_args=default_args,
    description='A simple tutorial DAG',
    schedule_interval=timedelta(days=1),
)

# 태스크로 사용할 Python 함수들 정의
def task1_function():
    print("This is Task 1")

def task2_function():
    print("This is Task 2")

def task3_function():
    print("This is Task 3")

# 태스크 생성
task1 = PythonOperator(
    task_id='task1',
    python_callable=task1_function,
    dag=dag,
)

task2 = PythonOperator(
    task_id='task2',
    python_callable=task2_function,
    dag=dag,
)

task3 = PythonOperator(
    task_id='task3',
    python_callable=task3_function,
    dag=dag,
)

# 태스크 의존성 설정
task1 >> [task2, task3]

이 코드는 간단한 DAG를 정의해요. 세 개의 태스크가 있고, task1이 완료된 후 task2와 task3이 병렬로 실행되는 구조예요. 마치 재능넷에서 기초 과정을 배운 후 두 가지 심화 과정을 동시에 학습하는 것과 비슷하죠? 😉

4.2 DAG 구조 시각화하기 🖼️

방금 만든 DAG의 구조를 시각화하면 이렇게 보일 거예요:

4.3 DAG 실행하기 ▶️

이제 DAG를 만들었으니 실행해볼 차례예요. Airflow CLI를 사용해 DAG를 트리거할 수 있어요:

airflow dags trigger my_first_dag

이 명령어를 실행하면 Airflow가 my_first_dag를 실행하기 시작할 거예요. Airflow UI에서 DAG의 실행 상태를 확인할 수 있답니다.

4.4 DAG 모니터링하기 👀

Airflow UI (http://localhost:8080)에 접속하면 DAG의 실행 상태를 모니터링할 수 있어요. UI에서는 다음과 같은 정보를 확인할 수 있죠:

• DAG의 전체적인 실행 상태
• 각 태스크의 상태 (성공, 실패, 실행 중 등)
• 태스크 로그
• DAG의 그래프 뷰와 트리 뷰

이렇게 모니터링을 하면 마치 재능넷에서 자신의 학습 진도를 체크하는 것처럼, DAG의 진행 상황을 한눈에 파악할 수 있어요. 😊

4.5 DAG 개선하기 🔧

첫 번째 DAG를 만들고 실행해봤어요. 하지만 이게 끝이 아니에요! DAG를 계속해서 개선하고 발전시킬 수 있답니다. 예를 들어:

• 더 복잡한 태스크 의존성 추가하기
• 다양한 Operator 사용해보기 (예: BashOperator, SQLOperator 등)
• 조건부 실행 로직 추가하기
• 에러 처리와 재시도 로직 개선하기

이렇게 계속해서 DAG를 개선하다 보면, 복잡한 데이터 파이프라인도 쉽게 만들 수 있게 될 거예요. 마치 재능넷에서 꾸준히 학습하며 실력을 키우는 것처럼 말이죠! 🚀

지금까지 첫 번째 DAG를 만들고 실행해보는 과정을 살펴봤어요. 이제 Airflow의 기본적인 사용법을 익혔으니, 더 복잡하고 실용적인 데이터 파이프라인을 만들 준비가 된 거예요. 다음 섹션에서는 실제 데이터 처리 작업을 포함한 더 복잡한 DAG를 만들어볼 거예요. 준비되셨나요? 그럼 계속해서 Airflow의 세계를 탐험해볼까요? 😃

5. 실용적인 DAG 만들기: 데이터 처리 파이프라인 🔄

자, 이제 기본적인 DAG 작성법을 익혔으니 조금 더 실용적인 DAG를 만들어볼까요? 이번에는 실제 데이터를 처리하는 파이프라인을 만들어볼 거예요. 마치 재능넷에서 배운 기술을 실제 프로젝트에 적용하는 것처럼 말이죠! 😊

5.1 시나리오 설정 📊

다음과 같은 시나리오를 가정해볼게요:

이 시나리오를 Airflow DAG로 구현해볼까요?

5.2 필요한 라이브러리 설치 📚

먼저, 필요한 라이브러리들을 설치해야 해요:

pip install pandas sqlalchemy psycopg2-binary

5.3 DAG 코드 작성하기 💻

이제 DAG 코드를 작성해볼게요. 이 코드를 sales_data_pipeline.py라는 이름으로 AIRFLOW_HOME/dags 디렉토리에 저장해주세요.

import pandas as pd
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.postgres_operator import PostgresOperator
from airflow.operators.email_operator import EmailOperator
from sqlalchemy import create_engine

# DAG 기본 인자 설정
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email': ['your_email@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG 정의
dag = DAG(
    'sales_data_pipeline',
    default_args=default_args,
    description='A pipeline to process daily sales data',
    schedule_interval=timedelta(days=1),
)

# 데이터 처리 함수
def process_sales_data():
    # CSV 파일 읽기
    df = pd.read_csv('/path/to/sales_data.csv')
    
    # 데이터 처리 (예: 날짜 형식 변경, 결측치 처리 등)
    df['date'] = pd.to_datetime(df['date'])
    df['total_sales'] = df['quantity'] * df['price']
    
    # 처리된 데이터를 데이터베이스에 저장
    engine = create_engine('postgresql://username:password@localhost:5432/sales_db')
    df.to_sql('daily_sales', engine, if_exists='replace', index=False)

# 분석 함수
def analyze_sales_data():
    engine = create_engine('postgresql://username:password@localhost:5432/sales_db')
    df = pd.read_sql('SELECT * FROM daily_sales', engine)
    
    total_sales = df['total_sales'].sum()
    avg_sales = df['total_sales'].mean()
    
    with open('/tmp/sales_report.txt', 'w') as f:
        f.write(f"Total Sales: {total_sales}\n")
        f.write(f"Average Sales: {avg_sales}\n")

# 태스크 정의
t1 = PythonOperator(
    task_id='process_sales_data',
    python_callable=process_sales_data,
    dag=dag,
)

t2 = PostgresOperator(
    task_id='create_summary_table',
    postgres_conn_id='postgres_default',
    sql="""
    CREATE TABLE IF NOT EXISTS sales_summary AS (
        SELECT date, SUM(total_sales) as daily_total
        FROM daily_sales
        GROUP BY date
    )
    """,
    dag=dag,
)

t3 = PythonOperator(
    task_id='analyze_sales_data',
    python_callable=analyze_sales_data,
    dag=dag,
)

t4 = EmailOperator(
    task_id='send_email',
    to='manager@example.com',
    subject='Daily Sales Report',
    html_content='Please find attached the daily sales report.',
    files=['/tmp/sales_report.txt'],
    dag=dag,
)

# 태스크 의존성 설정
t1 >> t2 >> t3 >> t4

이 DAG는 다음과 같은 단계로 구성되어 있어요:

1. CSV 파일에서 판매 데이터를 읽고 처리한 후 데이터베이스에 저장
2. 데이터베이스에 요약 테이블 생성
3. 판매 데이터 분석 수행
4. 분석 결과를 이메일로 전송

5.4 DAG 구조 시각화 🖼️

이 DAG의 구조를 시각화하면 다음과 같아요:

5.5 주의사항 및 개선 포인트 🚧

이 DAG를 실제로 사용할 때는 몇 가지 주의해야 할 점이 있어요:

• 보안: 데이터베이스 연결 정보는 Airflow의 Connection 기능을 사용해 안전하게 관리해야 해요.
• 에러 처리: 각 태스크에 적절한 에러 처리 로직을 추가하면 좋아요.
• 확장성: 데이터 양이 많아지면 Pandas 대신 Spark 같은 분산 처리 도구를 고려해볼 수 있어요.
• 모니터링: 중요한 지표들을 로깅하고 모니터링하는 것이 좋아요.

이렇게 만든 DAG는 실제 비즈니스 환경에서 매우 유용하게 사용될 수 있어요. 마치 재능넷에서 배운 기술을 실제 프로젝트에 적용하는 것처럼, 여러분도 이런 DAG를 만들어 실제 데이터 처리 작업을 자동화할 수 있답니다! 🚀

지금까지 실용적인 데이터 처리 파이프라인 DAG를 만들어봤어요. 이런 식으로 Airflow를 활용하면 복잡한 데이터 워크플로우도 효율적으로 관리할 수 있어요. 다음 섹션에서는 Airflow의 고급 기능들과 베스트 프랙티스에 대해 알아볼 거예요. 계속해서 Airflow 마스터의 길로 나아가볼까요? 😊

6. Airflow 고급 기능 및 베스트 프랙티스 🏆

자, 이제 Airflow의 기본적인 사용법과 실용적인 DAG 작성법을 배웠어요. 하지만 Airflow의 진정한 힘은 그 고급 기능들에 있답니다. 마치 재능넷에서 고급 과정을 수강하는 것처럼, 우리도 Airflow의 고급 기능들을 살펴보고 베스트 프랙티스를 알아볼까요? 🚀

6.1 동적 DAG 생성 🔄

때로는 비슷한 구조의 DAG를 여러 개 만들어야 할 때가 있어요. 이럴 때 동적 DAG 생성 기능을 사용하면 편리해요.

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

def create_dag(dag_id, schedule, default_args):
    def hello_world_py(*args):
        print('Hello World')
        print('This is DAG: {}'.format(dag_id))

    dag = DAG(dag_id, schedule_interval=schedule, default_args=default_args)

    with dag:
        t1 = PythonOperator(
            task_id='hello_world',
            python_callable=hello_world_py,
            dag=dag)

    return dag

# 여러 DAG 생성
for i in range(1, 4):
    dag_id = 'hello_world_{}'.format(i)
    default_args = {
        'owner': 'airflow',
        'start_date': datetime(2021, 1, 1),
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
    }
    schedule = '@daily'
    globals()[dag_id] = create_dag(dag_id, schedule, default_args)

이 코드는 비슷한 구조의 DAG를 여러 개 생성해요. 마치 재능넷에서 비슷한 주제의 강의를 여러 개 만드는 것과 같죠? 😉

6.2 Branching 🌿

때로는 조건에 따라 다른 태스크를 실행해야 할 때가 있어요. 이럴 때 BranchPythonOperator를 사용할 수 있어요.

from airflow.operators.python_operator import BranchPythonOperator

def branch_func(**kwargs):
    ti = kwargs['ti']
    xcom_value = ti.xcom_pull(task_ids='get_data')
    if xcom_value > 5:
        return 'task_a'
    else:
        return 'task_b'

branch_op = BranchPythonOperator(
    task_id='branch_task',
    python_callable=branch_func,
    provide_context=True,
    dag=dag)

task_a = DummyOperator(task_id='task_a', dag=dag)
task_b = DummyOperator(task_id='task_b', dag=dag)

branch_op >> [task_a, task_b]

이렇게 하면 조건에 따라 다른 태스크를 실행할 수 있어요. 마치 재능넷에서 학습자의 수준에 따라 다른 강의를 추천하는 것과 비슷하죠? 🌟

6.3 SubDAGs 🧩

복잡한 워크플로우를 관리하기 쉽게 만들기 위해 SubDAG를 사용할 수 있어요.

from airflow.operators.subdag_operator import SubDagOperator
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator

def subdag(parent_dag_name, child_dag_name, args):
    dag_subdag = DAG(
        dag_id=f'{parent_dag_name}.{child_dag_name}',
        default_args=args,
        schedule_interval="@daily",
    )
    for i in range(5):
        DummyOperator(
            task_id=f'{child_dag_name}-task-{i + 1}',
            default_args=args,
            dag=dag_subdag,
        )
    return dag_subdag

subdag_1 = SubDagOperator(
    task_id='subdag-1',
    subdag=subdag('parent_dag', 'subdag-1', default_args),
    default_args=default_args,
    dag=dag,
)

SubDAG를 사용하면 복잡한 워크플로우를 모듈화할 수 있어요. 마치 재능넷에서 큰 주제를 작은 단원으로 나누는 것과 같죠! 📚

6.4 SLAs와 Callbacks ⏰

SLA(Service Level Agreement)를 설정하고 콜백 함수를 사용하면 DAG의 실행을 더 잘 제어할 수 있어요.

def sla_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
    print("SLA was missed on DAG {0}s by task id {1}s".format(dag.dag_id, task_list))

def failure_callback(context):
    print("Task failed. Dag Id: {0}. Task Id: {1}".format(context['dag'].dag_id, context['task'].task_id))

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': days_ago(2),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    'sla': timedelta(hours=2),
    'on_failure_callback': failure_callback
}

dag = DAG(
    'example_sla_dag',
    default_args=default_args,
    description='A simple tutorial DAG',
    schedule_interval=timedelta(days=1),
    sla_miss_callback=sla_callback,
)

이렇게 SLA와 콜백을 설정하면 DAG의 실행을 더 잘 모니터링하고 관리할 수 있어요. 마치 재능넷에서 학습 진도와 성과를 체크하는 것과 비슷하죠? 📊

6.5 베스트 프랙티스 🌟

마지막으로, Airflow를 사용할 때 알아두면 좋은 베스트 프랙티스들을 정리해볼게요:

• 멱등성 유지: DAG를 여러 번 실행해도 같은 결과가 나오도록 설계하세요.
• 태스크 세분화: 큰 태스크를 작은 단위로 나누면 재실행과 디버깅이 쉬워져요.
• 변수 사용: 하드코딩된 값 대신 Airflow Variables를 사용하세요.
• 코드 재사용: 공 통 로직은 Python 모듈로 분리하여 재사용성을 높이세요.
• 로깅 활용: 중요한 정보는 로그로 남겨 디버깅과 모니터링을 용이하게 하세요.
• 테스트 작성: DAG와 태스크에 대한 단위 테스트를 작성하여 안정성을 높이세요.
• 버전 관리: DAG 코드를 버전 관리 시스템(예: Git)으로 관리하세요.

이러한 베스트 프랙티스를 따르면 Airflow를 더욱 효과적으로 사용할 수 있어요. 마치 재능넷에서 학습 팁을 따라 더 효율적으로 공부하는 것과 같죠? 😊

이렇게 Airflow의 고급 기능들과 베스트 프랙티스를 알아봤어요. 이 기능들을 잘 활용하면 더욱 강력하고 유연한 데이터 파이프라인을 구축할 수 있답니다. 마치 재능넷에서 고급 과정을 수료하고 나면 더 복잡한 프로젝트를 수행할 수 있는 것처럼 말이에요! 🎓

Airflow는 정말 강력한 도구지만, 동시에 계속해서 발전하고 있는 기술이에요. 새로운 기능과 업데이트를 계속 학습하고 적용하는 것이 중요해요. 마치 재능넷에서 새로운 강의가 올라올 때마다 학습하는 것처럼 말이죠! 🚀

여러분, 이제 Airflow의 기본부터 고급 기능까지 전반적인 내용을 살펴봤어요. 이 지식을 바탕으로 여러분만의 강력한 데이터 파이프라인을 구축해보세요. 데이터 과학의 세계에서 여러분의 재능이 빛을 발할 거예요. 마치 재능넷에서 여러분의 재능이 빛나는 것처럼 말이죠! 😊

Airflow와 함께하는 여정이 즐거우셨기를 바랍니다. 앞으로도 계속해서 학습하고 성장하세요. 여러분의 데이터 과학 여정을 응원합니다! 🎉