콘텐츠로 이동

Airflow 오퍼레이터

개요

DAG가 순서를 그린다면, 실제로 "일을 하는" 단위는 Task예요. 그런데 매번 Task를 처음부터 짜면 비효율적이죠. Airflow는 Operator라는 재사용 가능한 Task 템플릿을 제공해요. 공식 문서는 Operator를 "미리 정의된 Task의 템플릿으로, 로직이 다 짜여 있어서 인자만 넣으면 되는 것"이라고 설명해요. bash 명령을 실행하는 BashOperator, Python 함수를 호출하는 PythonOperator처럼, 흔한 작업 유형은 거의 템플릿으로 준비돼 있어요.

핵심 개념

Operator vs Task

Airflow 코드 안에서는 Task와 Operator를 섞어 쓰기도 해요. 개념적으로는 이렇게 구분돼요. Task는 DAG의 "실행 단위"를 뜻하는 일반적인 말이고, Operator는 그런 Task를 찍어내는 재사용 가능한 템플릿이에요. 예를 들어 BashOperator(task_id="run_backup", bash_command="...")는 bash 실행이라는 로직이 다 들어 있는 Operator를 하나의 Task로 만드는 거죠.

자주 쓰는 Operator

  • BashOperator — bash 명령을 실행해요. 셸 스크립트를 그대로 실행할 때 써요.
  • PythonOperator — 임의의 Python 함수를 호출해요.
  • @task 데코레이터 — Python 호출 가능(callable)을 Task로 바꿔주는 방식. 공식 문서는 템플릿 렌더링이 필요 없는 Python 작업엔 오히려 @task가 PythonOperator보다 권장된다고 해요.

이 외에도 많은 core Operator가 있고, 설치되어 있지 않다면 커뮤니티 providers 패키지에서 찾을 수 있어요. HTTP 호출, DB 적재, 파일 전송 등 생태계가 넓어요.

Jinja 템플릿

Airflow의 강력한 기능 중 하나가 Jinja 템플릿이에요. 템플릿으로 표시된 Operator 인자는 실행 시점에 값으로 치환돼요. 예를 들어 데이터 구간의 시작일을 환경 변수로 넘기고 싶다면 이렇게 해요.

date = "{{ ds }}"
t = BashOperator(
    task_id="test_env",
    bash_command="/tmp/test.sh",
    env={"DATA_INTERVAL_START": date},
)

{{ ds }}는 데이터 구간의 시작일(YYY-MM-DD)로 치환돼요. 템플릿 필드는 실행 직전에 렌더링되므로, DAG 코드에 값을 하드코딩하지 않아도 매 실행 시점에 맞게 변해요.

주의할 점은 템플릿과 Python f-string이 둘 다 {}를 쓴다 보니 충돌할 수 있다는 거예요. f-string 안에서 Jinja 표현식을 그대로 남기려면 중괄호를 네 개({{{{ ds }}}})로 써서 이스케이프해야 해요.

파일 참조 템플릿

BashOperator처럼 template_ext에 특정 확장자를 가진 문자열은 파일 경로로 취급돼요. bash_command="script.sh"처럼 쓰면 script.sh 파일 내용을 명령으로 써요. 반대로 cat script.sh 같은 실제 명령을 쓰려고 하는데 템플릿이 파일로 오인하면 실패하니, literal("cat script.sh")로 감싸거나 template_ext를 비워서 방지할 수 있어요.

pre_execute / post_execute

Operator가 실행되기 전과 후에 호출되는 pre_execute·post_execute 메서드가 있어요. 전 단계에서 이 작업을 건너뛸지 판단하거나(AirflowSkipException으로 skip), 후 단계에서 임시 파일을 정리하는 용도로 써요. 여러 작업 instance를 감싸는 setup/teardown 과는 다르다는 점을 구분해요.

실제 적용 (데이터스케쳐스)

Events 파이프라인에서 오퍼레이터 선택은 곧 "이 작업을 어떤 도구로 실행할지" 를 정하는 거예요. 예를 들어 원천에서 데이터를 가져오는 작업은 HTTP 호출 operator 또는 Python 함수(@task), 표준 스키마로 정리하는 무거운 변환은 Spark로 넘기는 operator, 결과를 창고에 적재하는 작업은 해당 DB operator를 쓰는 식이죠.

이때 헷갈리기 쉬운 건 "모든 걸 Python으로 다 짜려는 것"이에요. 이미 오퍼레이터로 준비된 작업(예: HTTP 호출, DB 적재)은 직접 구현하는 것보다 검증된 템플릿을 쓰는 게 안전해요. 또 템플릿({{ ds }})으로 데이터 구간을 넘기는 습관을 들이면, 각 Dag Run이 자기 데이터 구간을 올바르게 처리해요. 실제 무거운 계산은 Spark가 맡으므로(Spark), Airflow의 오퍼레이터는 그 실행을 호출·감독하는 역할에 집중해요.

더 알아보기