Airflow

데이터 파이프라인 소개

• 다음 주 날씨를 알려주는 쇼핑몰의 날씨 대시보드를 구축
  

• 파이프라인은 비교적 간단하지만 서로 다른 태스크로 구성

  • 태스크(task)간의 의존성을 명확하게 확인하는 방법 중 하나는 Dag를 이용
  • Dag는 방향성을 가지지만 비순환
  • 날씨 예보 가져오기 -> 예보 데이터 정제하기 -> 대시보드에서 데이터 전송

import airflow.utils.dates
from airflow import DAG
from airflow.operators.dummy import DummyOperator

dag = DAG(
    dag_id="01_umbrella",
    description="Umbrella example with DummyOperators.",
    start_date=airflow.utils.dates.days_ago(5),
    schedule_interval="@daily",
)

fetch_weather_forecast = DummyOperator(task_id="fetch_weather_forecast", dag=dag)
fetch_sales_data = DummyOperator(task_id="fetch_sales_data", dag=dag)
clean_forecast_data = DummyOperator(task_id="clean_forecast_data", dag=dag)
clean_sales_data = DummyOperator(task_id="clean_sales_data", dag=dag)
join_datasets = DummyOperator(task_id="join_datasets", dag=dag)
train_ml_model = DummyOperator(task_id="train_ml_model", dag=dag)
deploy_ml_model = DummyOperator(task_id="deploy_ml_model", dag=dag)

# Set dependencies between all tasks
fetch_weather_forecast >> clean_forecast_data
fetch_sales_data >> clean_sales_data
[clean_forecast_data, clean_sales_data] >> join_datasets
join_datasets >> train_ml_model >> deploy_ml_model