개요
2024.04.05-[Airflow] ExternalTaskSensor - 다른 dag/task 작업 완료 대기에서 다른 Dag나 Task의 실행을 대기하는 Sensor를 살펴보았는데, 확인해 보니 Airflow에서는 다양한 종류의 작업에 대한 Sensor를 구현할 수 있도록 Sensor 라이브러리를 제공하고 있었다.
이번 글에서는 Airflow에서의 Sensor 개념과 지원하고 있는 Sensor에 대해서 간단히 정리해두려고 한다.
Sensor
Airflow에서 Sensor란 현실 세계에서의 센서와 동일하게 어떠한 일이 발생할 떄까지 기다리도록 설계된 오퍼레이터이다. Sensor가 실행되면 특정 조건이 충족되는지 주기마다 확인하고, 조건이 충족되면 Task를 성공 상태로 표시하고 다음 작업을 실행되게 한다. 만약 조건이 충족되지 않으면 다음 주기 때 다시 조건이 충족되는지 확인한다.
Sensor를 적절하게 사용하면 이벤트 중심 Dag를 구성할 수 있다.
BaseSensorOperator
모든 Sensor는 BaseSensorOperator를 상속하며, 다음의 매개변수를 가질 수 있다.
- mode : Sensor 동작 방식
- poke : 기본값이다. poke 사이에는 sleep 상태를 유지하여 sensor의 전체 실행 시간 동안 worker의 슬롯을 점유한다. 실행 시간이 짧은 센서일 때 적절하다.
- reschedule : 조건을 만족하지 않았을 때 task가 worker 슬롯을 놓아주고 up_for_reschedule 상태를 유지하다가 다시 스케쥴링된다. 실행 시간이 길어질 센서일 때 적절하다.
- poke_interval : 조건을 확인하는 주기. 초 단위로 지정한다.
- exponential_backoff : True로 설정하면 이전 poke와 다음 poke 사이의 간격을 지수적으로 증가시킨다.
- timeout : task의 최대 실행 시간. 초 단위로 지정한다.
- soft_fail : True로 설정하면 Task Timeout 시 Failed가 아닌 Skipped로 표시된다.
지원 Sensor
Airflow에서는 다음과 같은 종류의 Sensor를 제공하고 있다. 많이 사용하는 것 위주로 적어둔다.
- PythonSensor : python_callable에 지정한 함수가 True를 반환할 때까지 대기한다. Airflow가 제공하고 있는 Sensor 외의 복잡한 조건을 구현할 때 적절한다.
- @task.sensor 데코레이터 : Python 함수를 BaseSensorOperator 객체로 변환하기 위해 사용한다.
- FileSensor : 특정 파일이 존재하면 True를 반환한다.
- DateTimeSensor : 지정한 시각이 지날 때까지 기다린다. Dag 내 다른 Task를 다른 시각에 실행시킬 때 유용하다.
- ExternalTaskSensor : 다른 Dag나 Task가 실행을 완료할 때까지 대기한다.
- HttpSensor : API 요청이 성공하는 것을 보장해야 할 때 사용하기 적절하다.
- SqlSensor : SQL 테이블 내 데이터가 존재할 때까지 대기한다.
- S3KeySensor : S3 버킷 내 Key가 나타날 때까지 대기한다. Key에 대한 head_object/list_objects_v2를 호출하여 확인한다.
이 외로, bash, TimeDeltaSensor, DayOfWeekSensor 등등이 존재한다.
참고 문서
https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/sensors.html#sensors
https://docs.astronomer.io/learn/what-is-a-sensor
https://airflow.apache.org/docs/apache-airflow/stable/release_notes.html