정선생의 데이터 팡팡

airflow 에서 logical_date 와 schedule 을 고려하면 월배치를 돌리려면 말일 기준의 logical_date 기준으로 배치가 돌아야 한다. 하지만 다 알고 있는것처럼 월말은 1월은 31일이요, 2월은 28 혹은 29일, 3월은 31일 등등 이런식이라 처리하기가 어렵다.제대로 처리하려면 Airflow 2.4.x 이상버전에서는 Timetables 을 직접 구현하는게 가장 베스트인걸로 보인다.하지만, 여기서는 AirflowSkipException 과 schedule 의 crontab 표현을 이용해 해결하는 방법을 알려주고자 한다. 쉽게 생각해서 crontab 에서 매월 28,29,30,31 에 스케쥴을 활성화 하고,월말을 계산해서 아닌날은 스케쥴을 넘기는 형태로 해결하는 방법이다. 해결방법..

airflow 2.x 로 올라가면서 ui 에서 타임존을 고려한 처리가 좋아지긴했는데, DAG 를 만들때 logical_date 를 이용해 값을 유도할때 우리가 의도한 값이 안나올때가 있다. 우선 airflow 의 환경변수기반으로 구성했다는 가정으로 타임존을 아래와 같이 구성하고export AIRFLOW__CORE__DEFAULT_TIMEZONE="Asia/Seoul"export AIRFLOW__WEBSERVER__DEFAULT_UI_TIMEZONE="Asia/Seoul" DAG 선언할때도 start_date 값에 타임존까지 잘 지정했다면 반은 성공한것이다. 근데 , 2024-01-02 00:00 가 되면 2024-01-01 00:00 의 logical_date 가 바인딩될것으로 기대되는데 이상하게{{ l..

airlfow 에서 한국시간으로 사용하기위해서 은근히 번거로운일이 많다, start_date 에 타임존을 지정하고, 설정에 타임존을 지정하더라도 의도한 값을 뽑아내려면 logical_date 를 타임존에 맞춰 변환후 사용해야 제대로 값을 추출할 수 있는 경우가 많다.근데 이걸 매번 DAG 코드에 넣는건 비효율적이기 때문에, Plugin 에 jinja template 으로 커스텀 필터를 추가해서 사용하는 방법을 제안한다. 커스텀 필터 추가하기참고로 airflow 2.9.x , 2.10.x 버전에서 동작 테스트 했었고, plugins 폴더에 넣어주면 사용가능하다.아래와 같이 커스텀 필터를 추가하면 jinja template 에서 쉽게 변환이 가능하다.from airflow.plugins_manager imp..

월별 지표를 만들기위해 말일에 스케쥴을 어떻게 할지 고민하다가, crontab 표현으로 28~31 활성화 하고, 체크로직을 넣어 실행을 제한하는식으로 접근을 했는데 마지막날이 스케쥴 안되는 문제가 발생했다.  예를 들어, 아래와 같이 스케쥴을 걸면 10/28 , 10/29, 10/30, 10/31 이 스케쥴이 활성화 되면서 실행되길 기대했는데...31일 스케쥴이 활성화가 안되는 상황이다.with DAG( dag_id="sample_dag", start_date=datetime.datetime(2024, 10, 1, tzinfo=pendulum.timezone("Asia/Seoul")), schedule_interval="0 0 28-31 * *", # 매월 말일 실행 max_act..

airflow 2.x 부터 롤기반으로 권한관리가 일어나고, 이에 따른 퍼미션 권한이 존재한다.그런데 이게 어떤걸 의미하는건지 잘 정리가 안되서 간단히 정리하고자한다. 1. can read on DAGsDAG 를 볼 수 있는지 여부를 의미한다.리스트에 나와서 어떤 DAGS 가 있는지 조회된다의 의미일뿐 실행이나 해제같은 동작은 안된다. (진짜 목록에만 보일뿐 할 수 있는게 없음)  DAG 별로 따로 수동으로 관리하고, 기본적으로 조회가 안되게 하려면 이 옵션을 제거해주도록 하자.그러면 아래와 같이 기본적으로는 DAGS 의 목록이 조회되지 않는다.특정 DAG 만 조회 가능하게 하려면 "can read on DAG: " 형태로 부여하는것도 가능하다. 2. can edit on DAGsDAG 의 목록이 보이고,..

airflow 의 best practices 글을 읽어보면, Variable.get(키) 형태로 직접 값을 가져오지 않고, 바인딩처리해서 {{var.value.키}} 표현해서 사용하는것을 기본 가이드로 하고 있다. 그 이유는 일단 DAG 를 구성할때 top level code 에 관련된 로직이 존재한다면 DAG 실행뿐 아니라, 로딩되는 시점에서도 그 코드가 동작되기 때문에 성능에 문제가 될 수 있다. https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html#airflow-variables from airflow.models import Variable # Bad example foo_var = Variable.get("foo")..

airflow 로 스케쥴 관리를 하고 있는데, logical_date 가 1초 밀리는 희안한 일이 일어났다. logical_date 의 경우 RUN_ID 값에 시간값이 붙어서 쉽게 인지가 가능한데, 정상적일땐 아래와 같이 RUN_ID 값이 00 으로 딱 떨어졌는데, 어느순간 다음과 같이 RUN_ID 의 값이 00 으로 딱 안떨어지고, 뒷 단위가 조금씩 밀리는 현상이 발견되었다. 구분 정상일때 비정상일때 RUN_ID scheduled__2003-12-07T20:00:00+00:00 scheduled__2003-12-07T21:00:01.0099+00:00 이게 문제가 되는 이유는 ExternalTaskSensor 의 경우는 앞쪽 DAG 의 의존성을 체크할때 logical_date 가 같은 이력을 참조하기 ..

airflow 에서는 code 탭에서 dag 를 구성한 python 파일을 볼 수 있다. 그런데, 이 파일과 dag 값이 다르게 로딩되서 한참 삽질한 경험을 공유하고자 한다. 결론부터 말하면 dag 를 선언한 파일에서 다른 dag 의 파일을 import 하면서 영향을 받았다. 원인 확인방법 airflow 에서는 dag 파일을 읽고, dag bag 에 담아서 관리된다. 문제는 이때 관련된 정보가 잘못 인지되었던 문제였다. dag 로딩이 잘되었는지는 airflow dags list 명령으로 확인이 가능하다. 원래 primitive.py, hour.py, day.py 3개의 파일이 따로 존재하고, dag 도 파일별로 따로 선언했는데 아래와 같이 primitive.py 경로가 아닌 hour.py 로 잘못잡히는걸..