airflow 사용기

11월에는 워크플로우 배치 실행을 위해 airflow를 사용해 보았습니다.
  

Airflow의 주요 기능은 유연한 파이썬 프레임워크를 사용해 쉽게 데이터 파이프라인을 구축할 수 있게 해주며, 최신 기술 환경에서 접하게 되는 서로 다른 기술들을 연결할 수 있는 다양한 빌딩 블록을 제공하는 것입니다.
- 바스 하렌슬락, Data Pipelines with Apache Airflow

  
배치 실행을 해야 하는 워크플로우는 원천 데이터 입력(DB, HDFS, FTP) - 변환 - 출력(DB, HDFS) 으로 구성되어 있으며, 변환 작업은 NULL값 처리, 이상치 처리, 컬럼 생성 등 학습 데이터로 사용할 수 있는 일련의 작업이 속해있습니다. 각각의 task는 pyspark를 사용한 task 로 구성되어 있습니다.
  
기존에 airflow를 사용하지 않고 배치를 구성할 때에는 전체 작업을 단일 스크립트로 모놀리식으로 구성되어 있었습니다. 단일 스크립트 안에서 여러 작업을 실행할 때에는 전역적으로 선언된 spark session을 사용하여 데이터 I/O, 변환 작업을 수행할 수 있었습니다. 따라서 작업간 spark viewtable을 사용하여 처리 하는 것이 가능했습니다. 예를 들어 데이터 입력을 받아 A라는 뷰 테이블을 생성하고 다음 단계에서는 A테이블을 사용하여 변환 단계를 거쳐 B 테이블을 생성하고, 이후 단계에서는 B 테이블을 사용하여 C 테이블을 이용하는 것과 같이 이전 단계의 결과 뷰테이블을 사용하는 방식이었습니다. 
  
airflow를 적용하여 배치 워크플로우를 생성하면서 이런 어려움을 접했습니다.
  
" 각각의 변환 단계를 개별 task 로 생성하였을 때 이전 단계에서 viewtable을 생성하더라도 다음 task 에서 viewtable 조회시 테이블이 없다는 에러가 발생 "
  
이러한 문제는 task 간 spark session 공유가 되지 않아서 발생하였습니다. 변환하는 작업을 하나의 task 안에 모두 모아서 사용한다면 에러는 발생하지 않지만 기존의 방식과 크게 다르지 않았습니다. 기존의 스크립트를 변경하고 테스트하는 데에 드는 비용을 감수할 만한지도 의문이었습니다. 각각의 분리된 task로 관리하기 위해서는 각 단계별로 중간 데이터를 저장하고, 저장한 데이터를 다음 단계에서 다시 가져와서 사용하는 방식을 택해야 했습니다. 이 방식은 Disk I/O에 따른 오버헤드가 발생하는 점이 있었습니다.
  
어떠한 방법을 택하면 좋을지 고민을 하다가, 현재 진행 중인 dask 스터디 팀원분들께 의견을 구했습니다. 의견들을 종합해 보니 아래와 같이 정리되었습니다.
  

[task를 나누는 단위] 1. 각각의 단계 수행 시간이 짧다면 spark 작업 전체를 하나의 task로 만드는 편이 좋다. 2. 각각의 단계 수행시간이 길거나 비즈니스 로직 상 중간 데이터가 필요한 경우가 있다. 이러한 경우에는 각 단계를 별도의 task로 분리한다. task 수행의 결과도 단계별로 적재한다.    [task를 분리하여 관리하는 이유] - 중간에 실패한 task를 재실행하기 위해서 - task 별 중간 데이터를 남기는 이유는 비즈니스 로직이나 스키마 변경 시 대응하기 위함이고, 에러 대응에도 활용한다.    [태스크간 데이터 전달] - S3나 하둡 등 스토리지를 이용하여 처리하는 방식을 사용할 수 있다. - 중간 데이터를 테이블화해서 spark sql로 처리하기도 함. -  Disk I/O(ex- parquet로 떨구고 다시 로드)에 따른 추가적인 부하가 존재하지만, 단계별 처리 시간이 긴 경우에는 중간 파일을 저장해서 task 별로 나누는 것이 추후 운영에 편리할 수 있음.

  
비즈니스에 따라 중간 데이터가 필요한 경우는 어떤 것이 있을까 생각해 봤는데, 매출 데이터 예시를 생각해 보았습니다. 매출 raw data 추출 -> 월별 판매량 합계 계산 -> 월별 고객 수 데이터와 조인 -> 월별 1인당 평균 구매액 output 이러한 워크플로우 예시가 있다고 하면, 특정 단계의 데이터만 필요하거나 원 데이터만을 필요로 하는 경우가 있지 않을까 생각해보았습니다. 
  
11월에는 airflow를 사용해서 워크플로우를 작성하는 업무를 주로 수행했습니다. 기존에 작성되어 있는 스크립트를 task 단위로 나누어서 작성하고 실행해볼 수 있었습니다. airflow ui를 통해 단계별로 실행 상태를 확인할 수 있는 점이 편리하게 느껴졌습니다. task를 나누는 부분에서는 단계별 작업 소요 시간과 중간 데이터의 필요 여부에 따라 task의 분리 여부를 결정할 수 있다는 점을 배웠습니다!