개요 에어플로우를 LocalExecutor를 사용해서 하나의 인스턴스에 실행시키고 있었다. 그러나 인스턴스에 문제가 생기면 Airflow도 동작하지 않을 수 있는 문제가 있고, 인프라가 커졌을 때 확장이 힘들어서 Worker를 분리해야했다. airflow에서 공식적으로 제공하는 docker-compose에서는 celery를 사용하며, 보통 airflow를 구축할 때 celery를 사용하는 것 같다. 이렇게 클러스터로 구축을 하게 되면 리소스가 필요할 때 worker의 갯수만 늘리면 되고, 리소스가 필요 없을 때 불필요하게 낭비하지 않고 worker의 갯수를 줄이면 되기 때문에 좋다. 구축할 때는 AWS ECS, AWS EFS 환경에서 구축했다. 구축 과정 Dask Cluster를 활용하여 구축 시도 가장..

개요 Task는 에어플로우의 기본 실행 단위이다. Task에는 세가지 종류가 있다. 1. Operator: Bash, Python 등과 연결되어 있는 미리 정의된 작업 템플릿 2. Sensor: 외부 이벤트를 기다리는 연산자의 하위 클래스 3. TaskFlow: @task 데코레이터를 사용하여 패키지와 할 수 있는 기능(Airflow 2.0.0 신규 기능) 기본 개념 관계 작업간의 종속성을 설정하기 위해서 upstream, downstream을 사용할 수 있다. 연산자를 사용하여 종속성을 설정할 수 있다. a_task >> b_task >> [c_task, d_task] 작업은 기본적으로 자신의 정보를 전달하지 않으며 독립적으로 실행이 되지만, XCom을 사용하여 다른 태스크로 정보를 전달할 수 있다. ..

개요 에어플로우에서는 execution date 라는 개념이 있다. 과거에 실행하지 못한 작업을 그 시간대에 맞춰서 실행할 때 유용하다. 하지만 사용하면서도 계속 헷갈리는데 이번에 정리해보려고 한다. Execution Date란? 통계청에서 Parameter로 가져오는 날짜를 지정해야 하는 API가 있다고 생각해본다. 이 API를 사용하여 데이터를 수집하는 스크립트를 만들었다. 크론탭으로 이 스크립트를 매일 아침에 실행하게 하였는데 만약 3일 전(2021-06-14) 통계청 서버의 오류로 데이터를 수집하지 못한 것을 알아챘다면 스크립트의 매개변수 중 날짜를 2021-06-14로 설정한 후 다시 실행시켜야 한다. Airflow에는 Execution Date라는 개념이 있어서 매일 데이터를 수집하도록(@da..

개요 Airflow는 복잡한 워크플로우를 프로그래밍 방식으로 작성해서, 스케줄링하고 모니터링할 수 있는 플랫폼이다. 데이터 파이프라인을 이루고 있는 ETL 스크립트들을 스케줄링 할 때 crontab, cloudwatch 등을 사용하는 곳이 많다. 그러나 스크립트들이 많아지고 서로에 대한 의존성이 생기게 되면 컨트롤하기 어렵고, 기존 작업이 실패했을 때 다시 스크립트를 실행하려면 로그를 확인하고 실행해야 하는 등의 문제점이 생긴다. 이러면 바로 복구할수도 없고, 어디서 잘못되었는지 확인하기도 어렵고, 의존성이 있는 스크립트가 잘못되었는데 그 다음 스크립트가 실행이 되버리는 등의 문제점이 발생할 수 있다. Airflow에는 서로에 대한 의존성을 표현할 수 있고, 스크립트가 실패했을 때 알람을 보내 확인하고 ..

개요 에어플로우에서 작업이 실패했을 때 알 수 있는 방법은 Airflow 인터페이스 웹에 들어가서 확인을 하거나 email을 연동하거나 slack을 연동하는 등의 방법이 있다. 현재는 연동이 되어있지 않아서 작업에 실패하면 직접 들어가서 확인을 해야했는데, 이러면 알람이 늦어져서 조치가 어려워진다. 그래서 지금은 현재 사용하는 Slack과 연동을 해보려고 한다. - 이 문서에서 사용하는 에어플로우 버전은 2.0.0이다. Slack으로 메세지 보내는 방법 Airflow Operator 중에 slack으로 메세지를 보내주는 오퍼레이터가 있다. 자세한 내용은 이 링크를 참조하면 된다. 1. Airflow가 설치되어 있는 서버에서 apache-airflow-providers-slack 을 설치한다. python..