Airflow - Sensor

Airflow - Sensor

센서는 N개의 Task 중 하나라도 구동완료시 의존하는 다른 Task 하나 이상을 실행시키는데 유용하다.

유스케이스로 한번 알아보자.

유스케이스

쿠컬팡리라는 쇼핑몰에서 여러 광고를 진열중이다.
이 중 광고가 4가지가 존재한다.
1번째 광고는 쿠컬팡리 구독서비스 가입 이벤트이다.
2번째 광고는 참치회사에서 할인 해주는 행사를 디스플레이한다.
3번째는 강아지 사료 회사에서 할인해주는 상품 디스플레이
4번째는 애플 신제품 오픈 안내 광고 안내이다.
 
4가지 광고를 조회하는 클라이언트의 IP와 시간의 결과값을 분석해야한다.
같은 IP에서 출발하여 광고를 조회하는 중복된 조회는 스킵한다.
마지막으로 해당 결과 값을 엑셀로 만든 후 마케팅팀과 사업팀에 메일로 전송해야하는 케이스다.
 
클라이언트의 IP, 시간 자료는 광고마다 정제되지않은 데이터로 결과가 제공되며, 데이터 제공 시간이 제각각이다.
 
1번 광고는 평균적으로 분석 결과 완료시간이 익일 00시에 바로 나오고
2번 광고는 익일 오전 02시에, 3번광고는 오전 익일 03시, 4번 광고는 익일 오전 05시에 완료된 결과를 이전에 분석했다.

유스케이스 분석

1. 4가지 광고를 조회하는 클라이언트의 IP와 시간의 결과값을 분석해야한다.
   해당 데이터를 각 시간별로 제공 되는 것을 확인했다.
   또한 각 광고 데이터 결과는 정제되지않은 데이터이다.
2. 같은 IP에서 출발하여 광고를 조회하는 중복된 조회는 스킵한다.
   필터링 기술을 넣은 Task로 해결할 수 있어보인다.
3. 마지막으로 해당 결과 값을 매일 09:00시에 구동하여 엑셀로 만든 후
   엑셀 파일에 3열의 데이터를 둔다.
   1열 - 광고 타입
   2열 - IP
   3열 - 조회 시간
4. 마케팅팀과 사업팀에 메일로 전송해야하는 케이스다.
   3번 Task가 종료되면 메일로 전송하는 모듈에 엑셀 데이터를 던지면 될 것 같다.

Task 구성해보기.

지금까지 배웠던 내용으로 만들어보기.

Task를 위 이미지 처럼 구성하면 될거같다.
그런데 유스케이스 처럼 각 광고마다 데이터가 들어오는 시간이 다 다르다.

광고 데이터가 들어오는 시간은 모두 다르지만 타임라인을 보면 이 모든 과정이 하나의 DAG로 들어가있다보니 데이터 최종 필터, 메일전송 Task가 09시에 시작되기 위해서 최초 들어오는 광고 데이터 1번과 9시간이나 차이가 난다.
즉 메일을 보내기 위해서 1번 광고 데이터가 9시간동안 대기상태로 존재해야하는데 이게 과연 옳을까.
이를 해결하기 위해서 센서 Task가 존재한다.

FileSensor, PythonSensor

FileSensor Operator는 일정 주기동안 파일을 감지하여 파일 존재 여부를 확인 후 실행하는 Operator이다.
airflow 예제 - https://airflow.apache.org/docs/apache-airflow/stable/_modules/airflow/example_dags/example_sensors.html
PythonSensor Operator는 일정주기동안 callable을 실행하여 결과가 true일 경우에 센서와 연결된 Task를 실행하는 Operator이다.
airflow 문서 - https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/sensors/python/index.html

FileSensor를 추가한 결과

센서마다 파일의 의존성을 정의하고 파일이 생성됨을 판단하였을 때 다음 Task인 필터링을 진행하고 데이터베이스에 저장한다.

DAG -> DAG Trigger

Dag -> Dag를 트리거하려면 TriggerDagRunOperator를 사용하자.
TriggerDagRunOperator - https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/operators/trigger_dagrun/index.html#airflow.operators.trigger_dagrun.TriggerDagRunOperator
TriggerDagRunOperator를 적용한 결과는 아래 그림과 같을 것이다.

이렇게 되면 이전에 DAG 내부에서 지속적으로 의존하는 상태가 아니라, 붙였다 땠다 할 수 있는 모듈과 같은 형식으로 유지할 수 있어 재사용성이 좋다.