초보자를 위한 Apache Airflow 3 가이드

I’m happy to translate the article for you, but I’ll need the full text you’d like translated. Could you please paste the content (or the portion you want translated) here? I’ll keep the source line exactly as you provided and preserve all formatting, code blocks, URLs, and technical terms.

데이터 오케스트레이션이란?

In DataOps (Data Operations), 오케스트레이션은 데이터 워크플로(예: ETL 파이프라인)를 관리하여 작업이 적절한 시점에 올바른 순서대로 실행되도록 보장하는 기본 시스템입니다.

DAG란 무엇인가?

A DAG (Directed Acyclic Graph) 은 실행될 모든 작업을 포함하는 모델입니다.

작업이란?

A task 은 DAG 내부의 단일 작업 단위입니다.

비유: DAG를 오케스트라 지휘자로, 작업들을 악기로 생각해 보세요.

Core Airflow Components

왜 크론 작업만 사용하지 않을까?

크론 작업은 알람 시계와 같습니다 – 종속성을 고려하지 않고 정해진 시간에 스크립트를 실행합니다.

예시:

• extract.py가 오전 12시에 예약됨
• transform.py가 오전 1시 30분에 예약됨

추출 작업에 40분이 걸리더라도, 크론은 여전히 오전 1시 30분에 변환 작업을 트리거하므로 데이터가 손상되거나 충돌이 발생할 수 있습니다.

반면에 Airflow는 프로젝트 매니저와 같습니다: 작업 종속성을 존중하고, 상위 작업이 성공했을 때만 하위 작업을 실행합니다.

Example: Defining a Simple DAG (Python API)

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

# Step 1: Define your Python function
def my_function():
    # Your logic here
    pass

# Step 2: Set default arguments
default_args = {
    'owner': 'your_name',
    'depends_on_past': False,          # 이전 DAG 실행을 기다리지 않음
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': False,
    'retries': 1,                       # 실패 시 한 번 재시도
    'retry_delay': timedelta(minutes=5)
}

# Step 3: Create DAG object
with DAG(
    dag_id='template_dag',                 # 고유 DAG 식별자
    default_args=default_args,
    description='Template for new DAGs',
    schedule_interval='@daily',            # 실행 빈도
    catchup=False,                         # 과거 날짜에 대해 실행하지 않음
    max_active_runs=1                      # 한 번에 하나의 인스턴스만 실행
) as dag:

    # Step 4: Define tasks
    task1 = PythonOperator(
        task_id='python_task',
        python_callable=my_function
    )

    task2 = BashOperator(
        task_id='bash_task',
        bash_command='echo "Hello World"'
    )

    # Step 5: Set dependencies
    task1 >> task2

이러한 명령은 오케스트레이션 엔진에 의해 해석되어 사용 가능한 리소스를 사용해 순차적으로 실행됩니다. 이것을 데이터 엔지니어들은 Workflow‑as‑Code 라고 부릅니다.

예시: TaskFlow API (데코레이터 기반)

import json
from airflow.decorators import dag, task
from pendulum import datetime

# 1. Define the DAG using the @dag decorator
@dag(
    start_date=datetime(2024, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["example", "taskflow"],
)
def taskflow_etl_pipeline():

    # 2. Extract: returns a dictionary
    @task()
    def extract():
        data_string = '{"1001": 30.5, "1002": 28.2, "1003": 31.1}'
        return json.loads(data_string)

    # 3. Transform: receives data from the upstream task
    @task()
    def transform(raw_data: dict):
        total_value = sum(raw_data.values())
        return {"total": total_value, "count": len(raw_data)}

    # 4. Load: final task to "load" or print the data
    @task()
    def load(processed_data: dict):
        print(f"Loading data: Total value is {processed_data['total']}")

    # 5. Define dependencies by calling the functions
    raw_data = extract()
    summary = transform(raw_data)
    load(summary)

# Instantiate the DAG
taskflow_etl_pipeline()

DAG 실행 확인 방법

• CLI:

  airflow dags list               # Shows parsed DAGs
  airflow dags list-import-errors # Shows syntax errors, if any

• Web UI: http://localhost:8080에 접속하여 DAG 상태와 로그를 확인하고, 실행을 수동으로 트리거합니다.

확장 가능한 워크플로우를 위한 모범 사례

1. 멱등성(Idempotency) – 동일한 실행 날짜에 대해 작업을 여러 번 실행해도 동일한 결과가 나와야 합니다.

2. 원자성(Atomicity) – 각 작업은 하나의 명확히 정의된 연산을 수행해야 합니다. 변환 단계가 실패하면 모든 원시 데이터를 다시 가져오는 대신 해당 작업만 재시도하면 됩니다.

   
   왼쪽 – 모놀리식; 오른쪽 – 모듈식.

3. 캡슐화(Encapsulation) – DAG 구조는 최상위 수준에서만 정의합니다. 무거운 데이터 처리, API 호출, DB 쿼리는 파일의 전역 범위에 배치하지 않아야 합니다. 스케줄러가 파일을 파싱할 때마다 해당 코드를 실행하게 되며, 이는 Airflow 인스턴스를 크래시시킬 수 있습니다.

요약

Apache Airflow는 처음에 위협적으로 보일 수 있지만, 본질적으로는 혼돈에 질서를 부여하도록 설계된 도구에 불과합니다. 오케스트레이션을 수용함으로써, 고립된 수동 실행 스크립트를 신뢰할 수 있는 자동화된 데이터 파이프라인으로 전환할 수 있습니다.

오케스트레이션을 즐기세요!

데이터 오케스트레이션

• 데이터 오케스트레이션은 데이터 파이프라인에 필수적이며, 데이터 작업이 올바른 순서와 적절한 시점에 실행되도록 보장합니다.

DAGs

• DAG는 청사진이며, 작업 및 종속성의 지도를 제공하여 작업이 순서대로 실행되지 않도록 보장합니다.

Airflow

• Airflow는 작업을 실행하고 모니터링하는 물류를 처리함으로써 무거운 작업을 수행하므로 논리 구현에 집중할 수 있습니다.

코드로 워크플로우

• 전통적인 연산자든 현대적인, 파이썬식 TaskFlow API를 사용하든, 복잡한 파이프라인을 정의할 유연성을 가집니다.