[#15] 스프린트 DA 트랙 24주차 위클리 페이퍼(DAG와 Task의 관계, Operator의 종류와 활용 사례)

위클리 페이퍼는 현재 훈련받고 있는 코드잇 스프린트 데이터 애널리스트 트랙에서 매주마다 훈련생 스스로 프로그래밍 언어, 데이터분석, 통계, 머신러닝 등 특정 주제에 대하여 심화 학습을 할 수 있도록 제출하는 과제입니다.

(매주 2~3가지 주제를 스스로 알아보고 학습하여 관련된 내용을 정리하여 후에 취업 활동 간에 경험할 수 있는 기술 면접을 대비함.)

---

23주차에 이어 이번 24주차 위클리 페이퍼의 내용을 소개하겠습니다.

 

 

이번 15번째 위클리 페이퍼 주제는

1. DAG와 Task의 관계에 대해 예시를 들어 설명해주세요.

---

2. 수업시간에서 배우지 않은 Operator들의 종류와 활용 사례를 설명해주세요.

---

1. DAG와 Task의 관계에 대해 예시를 들어 설명해주세요.

데이터 엔지니어링이나 워크플로 자동화에 대해 이야기할 때 빠지지 않는 개념 중 하나가 바로 DAG(Directed Acyclic Graph)라고 합니다. 이와 함께 DAG를 구성하는 핵심 요소인 Task도 함께 이해해야만 작업의 구조를 제대로 파악할 수 있습니다.
그래서 이에 대한 내용을 이해할 수 있게 DAG와 Task의 개념, 그리고 둘의 관계를 쉽게 풀어서 설명하겠습니다.

---

DAG(Directed Acyclic Graph)란?

 

DAG는 작업(Task)들 간의 흐름(Flow)을 나타내는 방향성이 있는 비순환 그래프입니다.
다시 말해, 작업들이 어떤 순서로 실행되어야 하는지 전체적인 설계도를 제공하는 역할을 하는 그래프입니다.

• 방향성(Directed): 작업 간의 순서가 명확히 정해져 있습니다. 예를 들어, Task A가 완료되어야 Task B가 실행될 수 있습니다.
• 비순환성(Acyclic): 작업 흐름 안에 순환(Loop)이 없습니다. 즉, 작업이 다시 이전 단계로 돌아가는 일이 없도록 설계됩니다.

DAG는 주로 복잡한 작업 흐름을 체계적으로 관리하고, 작업 간의 의존 관계를 시각화하기 위해 사용됩니다. 데이터 파이프라인, 워크플로 자동화, 머신러닝 모델 학습 파이프라인 등 다양한 분야에서 활용됩니다.

---

Task란?

Task는 DAG를 구성하는 개별 작업 단위로, 실제로 실행되는 작업의 최소 단위입니다.
Task는 데이터를 처리하거나, 특정한 연산을 수행하거나, 다른 시스템과 상호작용하는 등 다양한 기능을 수행할 수 있습니다.

• 독립성: Task는 독립적으로 정의되며, 특정 작업(Task A)이 완료된 후에 다른 작업(Task B)이 실행되도록 설정할 수 있습니다.
• 재사용성: 동일한 Task를 여러 DAG에서 재사용할 수 있습니다.
• 구체적인 역할: 예를 들어, 데이터를 가져오는 Task, 변환하는 Task, 데이터를 저장하는 Task 등이 있습니다.

---

DAG와 Task의 관계

DAG와 Task는 설계도와 실행 작업의 관계로 이해하면 쉽습니다.
DAG는 Task들의 흐름과 의존성을 설계하고, Task는 DAG에 따라 실행되는 구체적인 작업 단위입니다.

관계의 특징

1. 의존성 관리: DAG는 Task들 간의 실행 순서와 관계를 정의합니다.
   예를 들어, Task A → Task B → Task C의 형태로 순서를 지정할 수 있습니다.
   
   
2. 병렬 처리: Task 간에 의존 관계가 없는 경우, DAG는 Task를 병렬로 실행해 효율을 극대화합니다.
   예를 들어, Task A 이후에 Task B와 Task C가 동시에 실행될 수 있습니다.
   
   
3. 실패 시 제어: 특정 Task가 실패하면, DAG는 후속 작업의 실행을 중단하거나 오류를 처리하도록 설계할 수 있습니다.

---

예시: 데이터 파이프라인으로 이해하기

DAG 이름: daily_data_pipeline

Task 구성:

1. fetch_data: 외부 데이터 소스에서 데이터를 가져옵니다.
2. transform_data: 가져온 데이터를 정리하고 변환합니다.
3. load_data: 변환된 데이터를 데이터베이스에 적재합니다.

DAG 흐름:

fetch_data → transform_data → load_data

실행 과정:

• fetch_data가 성공적으로 완료되면 transform_data가 실행됩니다.
• transform_data가 완료된 이후에 load_data가 실행됩니다.

만약 fetch_data에서 오류가 발생하면, 이후 작업(transform_data, load_data)은 실행되지 않습니다.

병렬 처리의 예시:

만약 데이터 검증 작업(validate_data)이 transform_data와 독립적이라면, 두 작업은 동시에 실행될 수 있습니다.

---

DAG와 Task의 활용

DAG와 Task는 다음과 같은 다양한 분야에서 활용됩니다.

1. 데이터 파이프라인: 데이터를 수집, 변환, 적재하는 과정을 체계적으로 설계.
   예: Apache Airflow, Prefect.
2. 머신러닝 파이프라인: 모델 학습, 평가, 배포 과정을 자동화.
   예: Kubeflow, MLflow.
3. 분석 및 보고 자동화: 데이터를 분석하고 보고서를 생성 및 배포.
4. 클라우드 기반 데이터 처리: AWS, GCP, Azure 등 클라우드 환경에서 데이터 워크플로 관리.
   예: Google Cloud Composer

---

2. 수업시간에서 배우지 않은 Operator들의 종류와 활용 사례를 설명해주세요.

Airflow에서 사용하는 Operator들은 작업(task)을 정의하고 실행하는 데 중요한 요소로 작용하고 있습니다.
다양한 종류의 Operator가 있으며, 각각의 특성과 활용 사례를 이해하면 효과적으로 워크플로우를 설계할 수 있습니다.
아래는 주요 Operator의 종류와 활용 사례를 정리한 내용입니다.

---

1. BashOperator

• 설명: Bash 명령어나 스크립트를 실행하는 Operator입니다.
• 활용 사례:
  • 데이터 처리 전 간단한 환경 설정.
  • 쉘 스크립트를 실행하여 데이터를 다운로드하거나 이동.
  • 시스템 상태 점검 또는 로그 수집.

from airflow.operators.bash import BashOperator

task = BashOperator(
    task_id='print_date',
    bash_command='date',
)

 

---

2. PythonOperator

• 설명: Python 함수를 실행하는 Operator입니다.
• 활용 사례:
  • 데이터 처리 로직 구현.
  • API 호출 및 데이터 수집.
  • 사용자 정의 데이터 변환 작업.

from airflow.operators.python import PythonOperator

def my_function():
    print("Hello from PythonOperator!")

task = PythonOperator(
    task_id='python_task',
    python_callable=my_function,
)

 

---

3. DummyOperator

• 설명: 실행 작업 없이 워크플로 제어용으로 사용됩니다.
• 활용 사례:
  • 태스크 그룹의 시작/종료를 명시적으로 표시.
  • 특정 조건에 따라 태스크를 분기할 때 연결점으로 사용.

from airflow.operators.dummy import DummyOperator

task = DummyOperator(
    task_id='start_task',
)

 

---

4. BranchPythonOperator

• 설명: 조건에 따라 태스크를 분기 실행합니다.
• 활용 사례:
  • 특정 조건에 따라 서로 다른 데이터 처리를 실행.
  • API 응답에 따라 다른 태스크를 수행.

from airflow.operators.python import BranchPythonOperator

def branch_func():
    return 'task_1' if some_condition else 'task_2'

task = BranchPythonOperator(
    task_id='branch_task',
    python_callable=branch_func,
)

---

5. EmailOperator

• 설명: 이메일을 발송하는 Operator입니다.
• 활용 사례:
  • 워크플로 완료 또는 실패 알림.
  • 작업 결과 보고서 자동 발송.

from airflow.operators.email import EmailOperator

task = EmailOperator(
    task_id='send_email',
    to='example@example.com',
    subject='Airflow Task Complete',
    html_content='<p>The task is complete.</p>',
)

 

---

6. SqlOperator

• 설명: SQL 쿼리를 실행하는 Operator입니다.
• 활용 사례:
  • 데이터베이스에 데이터를 삽입, 업데이트, 삭제.
  • ETL 작업의 일부로 SQL 실행.
  • 데이터 검증 작업.

from airflow.providers.postgres.operators.postgres import PostgresOperator

task = PostgresOperator(
    task_id='run_sql',
    sql='SELECT * FROM my_table;',
    postgres_conn_id='my_postgres',
)

 

---

7. Sensor

• 설명: 특정 조건이 충족될 때까지 기다리는 Operator입니다.
  • 종류: FileSensor, TimeSensor, ExternalTaskSensor 등.
• 활용 사례:
  • 파일 생성 또는 데이터 업데이트를 감지.
  • 특정 시간까지 작업 대기.
  • 외부 워크플로의 완료 대기.

from airflow.sensors.filesystem import FileSensor

task = FileSensor(
    task_id='wait_for_file',
    filepath='/path/to/myfile',
    poke_interval=30,
    timeout=300,
)

 

---

8. SubDagOperator

• 설명: 워크플로 내에서 서브 DAG을 실행합니다.
• 활용 사례:
  • 복잡한 태스크를 서브 DAG으로 나누어 관리.
  • 반복적인 작업 그룹 실행.

SubDagOperator는 사용성이 떨어지고, TaskGroup 사용하는 것을 권장하고 있습니다.

---

9. DockerOperator

• 설명: Docker 컨테이너를 실행하는 Operator입니다.
• 활용 사례:
  • 독립적인 환경에서 데이터 처리 또는 머신러닝 모델 실행.
  • 종속성이 복잡한 애플리케이션 실행.

from airflow.providers.docker.operators.docker import DockerOperator

task = DockerOperator(
    task_id='run_docker',
    image='my-docker-image',
    command='python script.py',
    docker_url='unix://var/run/docker.sock',
)

---

10. KubernetesPodOperator

• 설명: Kubernetes 클러스터에서 Pod을 실행합니다.
• 활용 사례:
  • 클라우드 네이티브 환경에서 작업 실행.
  • 리소스 격리가 필요한 작업 처리.

---