Assets를 활용한 데이터 인식 스케줄링

Airflow로 데이터 파이프라인 구축하기

Volker Janz

Senior Developer Advocate at Astronomer

타이밍 문제

파이프라인 A는 API에서 원시 데이터를 불러옵니다
파이프라인 B는 해당 데이터로 대시보드를 생성합니다
둘 다 cron으로 실행하면, B가 A보다 먼저 완료될 수 있습니다
시간 버퍼를 추가하는 방법은 불안정합니다

타이밍 문제

Asset이란?

from airflow.sdk import Asset
sales_data = Asset("s3://bucket/sales/daily.parquet")

고유한 이름으로 식별되는 데이터에 대한 참조입니다
구체적인 데이터 엔티티를 나타낼 경우 URI를 asset에 연결할 수 있습니다
파일, 데이터베이스 테이블 등 모든 종류의 데이터가 될 수 있습니다

프로듀서와 컨슈머

태스크가 asset을 업데이트하면 asset 이벤트가 생성됩니다
asset 스케줄을 사용하면 asset 업데이트 시 DAG를 트리거할 수 있습니다

Asset 흐름

프로듀서: outlets로 신호 보내기

sales_data = Asset("s3://bucket/sales/daily.parquet")


@task(outlets=[sales_data])
def write_sales():
    # Write data to S3
    ...

성공 시 Airflow는 해당 asset이 업데이트되었음을 기록합니다
프로듀서는 컨슈머에 대해 알 필요가 없습니다

컨슈머: asset 기반 스케줄링

sales_data = Asset("s3://bucket/sales/daily.parquet")


@dag(schedule=[sales_data])
def build_dashboard():
    ...

cron 표현식 없이 asset 참조만 사용합니다
업데이트 후 Airflow가 컨슈머를 자동으로 트리거합니다

조건부 스케줄링

sales = Asset("s3://bucket/sales.parquet")
inventory = Asset("s3://bucket/inventory.parquet")


# Wait for BOTH assets to update
@dag(schedule=(sales & inventory))
def full_report(): ...


# Trigger when ANY asset updates
@dag(schedule=(sales | inventory))
def quick_refresh(): ...

CLI로 asset 확인하기

$ airflow assets list

name                                  | group    | uri
s3://data-lake/sales/daily.csv        | asset    | s3://data-lake/sales/daily.csv

$ airflow assets details --name "s3://data-lake/sales/daily.csv"

property_name    | property_value
=================+=====================                                         
name             | s3://data-lake/sales/daily.csv
                ...
updated_at       | 2026-04-14T10:00:50.820041Z

airflow assets list는 등록된 모든 asset을 표시합니다
airflow assets details는 updated_at을 포함한 메타데이터를 표시합니다

연습해 봅시다!

Airflow로 데이터 파이프라인 구축하기