빅데이터 기술 | Airflow로 데이터 파이프라인 생성하기 (ETL)

데이터 파이프라인 이란?

(a.k.a ETL(Extract-Transform-Load))

Data Source(log, api, product dataset,.. etc)에서 원하는 데이터를 Extract 해서 원하는 특정 format으로 Transform 하고 데이터를 원하는 곳(Data Warehouse,... etc)에 적제(Load)하는 일련의 과정 및 Architecture를 이야기한다.

(최근에는 순서를 바꿔 ELT 프로세스로 사용하기도 한다.)

 

멱등성이란?

데이터 파이프라인에서 가장 중요한 것은 재실행의 안전성이다. 즉, *멱등성(Idempotency)를 보장해야 한다. Airflow는 **Backfill이 쉽다.

*멱등성: 동일한 테스크를 여러 번 수행해도 동일한 결과가 되도록 하는 것 (도중에 오류가 발생해 재실행을 해도 중복이 발생하지 않아야 한다.)

**Backfill: 과거 데이터를 다시 채우는 과정

 

Airflow란?

Apache Airflow

워크 플로우 관리도구 중 하나로 데이터 파이프라인을 쉽게 해주는 프레임워크이다.

워크플로 관리 기술을 사용하면 스케줄러 기능뿐 아니라 데이터 파이프라인이 복잡해짐에 따라 크고 작은 장애가 발생하더라도 오류 처리와 다시 처리하기 위한 기능을 만들어 관리할 수 있다.

 

Airflow에 의한 워크플로는 여러 task(=operator)로 이루어진 DAG(Directed Acyclic Graph)의 형태로 정의한다.

 

예시

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

# config
dag = DAG(
    dag_id='helloword_dag',
    start_date=datetime(2022,12,19),
    catchup=False,
    tags=['example'],
    schedule_interval='0 2 * * *')  # crontab과 유사

def print_hello():
    print("hello!")
    return "hello!"

def print_goodbye():
    print("goodbye!")
    return "goodbye!"

# PythonOperator: 범용적
# 보통 task를 3개로 나눔 (ETL)
print_hello = PythonOperator(
    task_id='print_hello',
    python_callable=print_hello,
    dag=dag)

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

#Assign the order of the tasks in our DAG
print_hello >> print_goodbye

• 출처: '실리콘밸리에서 날아온 데이터 엔지니어링 스타터 키드' 스터디