Airflow

태스크는 트랜잭션처럼 - 멱등성(idempotency) 확보하기

Airflow의 태스크는 데이터베이스의 트랜잭션과 동등하게 취급해야 한다. 즉 태스크가 불완전한 결과를 만들어서는 안 된다 - 예를 들어 태스크가 실패했는데 HDFS나 S3에 일부만 쓰인 데이터가 남아있는 상황이 대표적인 안티패턴이다.

Airflow는 실패한 태스크를 재시도할 수 있으므로, 태스크는 매 재실행마다 동일한 결과를 내야 한다. 이를 위한 실천법은 세 가지다.

첫째, 재실행 시 INSERT 대신 UPSERT를 사용한다. INSERT는 재실행마다 중복 행을 만들 수 있다.

둘째, 특정 파티션을 지정해 읽고 쓴다. 태스크 안에서 "가장 최신 데이터"를 읽지 말아야 한다 - 재실행 사이에 누군가 입력 데이터를 갱신하면 실행마다 다른 결과가 나온다. 대신 ``data_interval_start``를 파티션 키로 사용해 입력을 읽고, S3/HDFS에 데이터를 쓸 때도 동일한 파티셔닝 방식을 따라야 한다.

셋째, ``datetime.now()``처럼 호출 시점에 따라 값이 달라지는 함수는 핵심 연산에 사용하지 않는다. 실행마다 다른 결과를 낳기 때문이다. 다만 임시 로그를 남기는 용도처럼 결과에 영향을 주지 않는 곳에는 사용해도 무방하다.

부가적인 팁으로, ``connection_id``나 S3 경로처럼 반복되는 파라미터는 태스크마다 선언하지 말고 ``default_args``에 정의하는 것이 좋다. 이는 오타 같은 실수를 줄여주고, 대부분의 커넥션 타입은 고유한 파라미터 이름(예: ``gcp_conn_id``)을 가지므로 ``default_args``에 한 번만 선언해도 해당 커넥션 타입을 쓰는 모든 오퍼레이터에 자동 적용된다.

태스크 삭제 시 주의사항

Dag에서 태스크를 삭제할 때는 주의가 필요하다. 태스크를 삭제하면 Graph View, Grid View 등에서 해당 태스크가 더 이상 보이지 않게 되어, 웹서버를 통해 그 태스크의 로그를 확인하기 어려워진다. 과거 실행 이력 조회가 중요하다면 태스크를 삭제하는 대신 새로운 Dag를 만드는 것이 바람직하다.

태스크 간 통신 - 로컬 파일 대신 XCom과 원격 스토리지

Kubernetes Executor나 Celery Executor를 사용하면 Dag의 태스크들은 서로 다른 서버에서 실행된다. 따라서 로컬 파일시스템에 파일이나 설정을 저장해서는 안 된다 - 다운스트림 태스크가 그 파일에 접근할 수 없는 다른 서버에서 실행될 가능성이 높기 때문이다(예: 한 태스크가 파일을 다운로드하고 다음 태스크가 그 파일을 처리하는 패턴). Local Executor를 쓰는 경우에도 디스크에 파일을 저장하면 재시도가 어려워질 수 있다 - 예를 들어 태스크에 필요한 설정 파일이 Dag 내 다른 태스크에 의해 삭제되는 경우다.

가능하다면 태스크 간 작은 메시지 전달에는 XCom을 사용하고, 태스크 간에 더 큰 데이터를 전달해야 한다면 S3/HDFS 같은 원격 스토리지를 사용하는 것이 좋은 방법이다. 예를 들어 처리된 데이터를 S3에 저장하는 태스크는 그 출력 데이터의 S3 경로를 XCom에 push하고, 다운스트림 태스크는 XCom에서 그 경로를 pull해 데이터를 읽는 데 사용할 수 있다.

태스크는 비밀번호나 토큰 같은 인증 파라미터를 태스크 내부에 저장해서도 안 된다. 가능하다면 Connections를 사용해 Airflow 백엔드에 데이터를 안전하게 저장하고, 고유한 connection id로 이를 조회해야 한다.