Airflow

Operator란 무엇인가

Operator는 개념적으로 미리 정의된 Task의 템플릿으로, Dag 안에서 선언적으로 정의할 수 있다. Airflow는 core나 사전 설치된 provider에 포함된 매우 방대한 오퍼레이터 집합을 갖고 있다. Core에서 인기 있는 오퍼레이터로는 bash 명령을 실행하는 ``BashOperator``, 임의의 Python 함수를 호출하는 ``PythonOperator``가 있으며, Jinja 템플릿 렌더링을 지원하지 않는 대신 임의의 Python 콜러블을 실행하고 싶다면 ``@task`` 데코레이터를 쓸 수 있다(``@task``는 템플릿 렌더링이 없는 ``PythonOperator``의 TaskFlow 대응 버전이다). 필요한 오퍼레이터가 기본 설치에 없다면 커뮤니티 provider 패키지에서 찾을 수 있는 경우가 많다 — ``EmailOperator``, ``HttpOperator``, ``SQLExecuteQueryOperator``, ``DockerOperator``, ``HiveOperator``, ``S3FileTransformOperator``, ``PrestoToMySqlOperator``, ``SlackAPIOperator`` 등이 그 예다.

Airflow 코드 내부에서는 Task와 Operator 개념이 종종 섞여 쓰이며 대부분 상호 교환 가능하다. 다만 일반적으로 'Task'라고 할 때는 Dag의 '실행 단위'라는 일반적인 개념을 가리키고, 'Operator'라고 할 때는 로직이 모두 미리 완성되어 있고 몇 가지 인자만 채우면 되는 재사용 가능한 사전 제작 Task 템플릿을 가리킨다.

Jinja 템플릿과 template_fields

Airflow는 Jinja 템플릿의 힘을 활용하며, 매크로와 결합하면 강력한 도구가 된다. 예를 들어 데이터 인터벌 시작 날짜를 Bash 스크립트에 환경변수로 넘기고 싶다면, ``BashOperator``의 ``env`` 파라미터가 Jinja 템플릿을 지원하므로 ``date = "{{ ds }}"``처럼 템플릿 변수를 쓸 수 있다. Jinja 대신 콜러블을 쓰고 싶다면, 그 콜러블은 ``context``와 ``jinja_env`` 두 개의 키워드 인자를 받아야 한다. ``context``는 현재 태스크 실행의 런타임 정보를 담은 읽기 전용 딕셔너리형 객체다. 템플릿 필드는 오퍼레이터의 ``pre_execute`` 함수가 호출되기 직전에 렌더링되며, 콜러블의 반환값은 다시 렌더링을 거치지 않으므로 콜러블 스스로 필요한 템플릿을 수동으로 렌더링해야 한다(``render_template()`` 메서드 사용).

'템플릿됨'으로 표시된 모든 파라미터에서 템플릿을 쓸 수 있으며, 중첩된 필드도 그 구조 안에서 템플릿됨으로 표시되어 있는 한 템플릿을 쓸 수 있다 — 각 클래스의 ``template_fields`` 클래스 변수(``Sequence[str]`` 타입)에 등록된 필드가 템플릿 치환 대상이 된다. 깊게 중첩된 필드도, 중간의 모든 필드가 각자 template_fields로 표시되어 있다면 치환될 수 있다. Dag 생성 시 ``jinja_environment_kwargs``로 Jinja ``Environment``에 커스텀 옵션을 전달할 수도 있다(예: 트레일링 뉴라인 유지).

일부 오퍼레이터는 ``template_ext``에 정의된 특정 접미사로 끝나는 문자열을 파일 참조로 간주한다 — 예를 들어 ``bash_command="script.sh"``라고 쓰면 그 경로의 파일 내용을 불러와 bash_command 값으로 사용한다. 이런 경로는 기본적으로 Dag 폴더 기준 상대 경로로 취급되며(Jinja의 기본 템플릿 검색 경로이기 때문), Dag의 ``template_searchpath`` 인자로 추가 경로를 등록할 수도 있다. 반대로 문자열을 파일 참조로 취급하지 않고 그대로 쓰고 싶다면 ``airflow.sdk.literal``로 감싸면 매크로와 파일 렌더링을 모두 비활성화할 수 있고(Airflow 2.8에서 추가), 또는 ``template_ext``를 오버라이드해 그 오퍼레이터 인스턴스에서 파일 참조 동작 자체를 끌 수도 있다.

Native Object 렌더링과 f-string 충돌

기본적으로 ``template_fields``의 모든 Jinja 템플릿은 문자열로 렌더링된다. 예를 들어 ``extract`` 태스크가 XCom으로 딕셔너리를 push했더라도, 이를 ``{{ ti.xcom_pull('extract') }}``로 참조하는 다운스트림 파라미터는 파이썬 dict가 아니라 그 dict를 문자열로 표현한 값을 받게 되어 ``.values()`` 같은 호출이 실패할 수 있다. 이를 해결하는 방법은 두 가지다: 콜러블 템플릿 필드를 써서 ``context["ti"].xcom_pull(...)``로 직접 실제 객체를 가져오거나, Dag에 ``render_template_as_native_obj=True``를 설정해 Jinja가 기본 ``SandboxedEnvironment`` 대신 ``NativeEnvironment``를 쓰도록 하는 것이다. ``NativeEnvironment``는 값을 Python 리터럴 규칙에 따라 렌더링하므로 리스트·딕셔너리·숫자·불리언을 만들어낼 수 있지만, ``"42"``처럼 숫자처럼 보이는 문자열이 정수 ``42``로 렌더링될 수도 있다는 부작용이 있다 — 값을 문자열로 유지해야 하는 태스크라면 기본 문자열 렌더링을 유지하거나, 콜러블 템플릿 필드를 쓰거나, 명시적으로 인용부호를 추가해야 한다.

Python f-string으로 템플릿 필드용 문자열을 구성할 때는 f-string과 Jinja가 똑같이 중괄호(``{}``)를 쓴다는 점에 주의해야 한다. f-string은 이중 중괄호(``{{``, ``}}``)를 리터럴 단일 중괄호를 위한 이스케이프로 해석하지만, Jinja는 이중 중괄호(``{{ variable }}``)를 변수 템플릿 표기로 쓴다. f-string 안에 ``{{ ds }}`` 같은 Jinja 템플릿 표현식을 그대로 남겨서 Airflow의 Jinja 엔진이 나중에 처리하게 하려면, f-string용으로 중괄호를 한 번 더 이스케이프해야 하므로 결과적으로 **네 개**의 중괄호(``f"echo Data interval start: {{{{ ds }}}}"``)를 써야 한다. 이렇게 하지 않으면 Dag 파싱 중 오류가 나거나, 템플릿이 예상대로 처리되지 않는 런타임 오류로 이어질 수 있다.

예약 키워드, pre_execute/post_execute, Setup/Teardown과의 차이

Airflow 2.2.0부터 ``params``라는 이름은 Dag 직렬화 과정에서 내부적으로 쓰이므로, 서드파티 오퍼레이터에서 이 이름을 필드명으로 재사용하면 안 된다. 업그레이드 후 ``AttributeError: 'str' object has no attribute '__module__'`` 오류가 난다면 오퍼레이터의 ``params`` 필드 이름을 바꿔야 한다.

``pre_execute``와 ``post_execute`` 메서드는 오퍼레이터 실행 전후에 각각 호출된다. 예를 들어 ``pre_execute``로 태스크가 실행되어야 할지 우아하게 판단할 수 있고(예: ``tasks_to_skip`` Variable에 현재 task_id가 있으면 ``AirflowSkipException``을 raise), ``post_execute``로는 오퍼레이터가 만든 임시 파일이나 디렉터리를 정리할 수 있다. 두 메서드 모두 태스크 인스턴스의 컨텍스트를 파라미터로 받는다.

``pre_execute``/``post_execute``는 개별 태스크 인스턴스 레벨에서 오퍼레이터 실행 전후에 호출되는 반면, Setup과 Teardown은 하나의 Dag Run 안에서 여러 태스크 인스턴스가 실행되기 전/후에 준비·정리 작업을 수행하는 특수한 태스크라는 점에서 서로 다른 개념이다.