Airflow

None 반환으로 매핑 항목 걸러내기

실제 태스크로 실행되는 매핑 태스크는 특정 입력에 대해 `None`을 반환함으로써 그 항목을 다운스트림으로 넘기지 않고 걸러낼 수 있다. 예를 들어 S3 버킷에서 특정 확장자를 가진 파일만 다른 버킷으로 복사하고 싶다면, `create_copy_kwargs` 태스크가 `.json`이나 `.yml`로 끝나지 않는 파일에 대해 `None`을 반환하도록 구현하면 된다. 이렇게 하면 `copy_files`는 `.json`/`.yml` 파일에 대해서만 확장되고 나머지는 무시된다.

map()으로 사전 변환하기

매핑을 위한 출력 데이터 형식을 변환하고 싶은 경우가 흔한데, 특히 출력 형식이 미리 정해져 있어 쉽게 바꿀 수 없는 비-TaskFlow 오퍼레이터에서 그렇다. 이럴 때 `.map()` 함수로 이런 변환을 간단히 수행할 수 있다.

`.map()`에 넘기는 콜러블(위 예의 `create_copy_kwargs`)은 태스크가 아니라 순수한 파이썬 함수여야 한다 - 이 변환은 다운스트림 태스크(예: `copy_files`)의 "전처리"의 일부로 취급되며, Dag 상의 독립된 태스크가 아니다. 이 콜러블은 항상 정확히 하나의 위치 인자를 받으며, 파이썬 내장 `map()`처럼 태스크 매핑에 쓰이는 iterable의 각 항목마다 한 번씩 호출된다.

이 콜러블은 다운스트림 태스크의 일부로 실행되므로 기존에 태스크 함수를 작성하던 모든 기법을 그대로 쓸 수 있다. 다만 항목을 스킵 처리하고 싶다면 `AirflowSkipException`을 raise해야 한다는 점에 유의해야 한다 - 매핑 태스크 자체의 필터링과 달리, `.map()` 안에서는 `None`을 반환해도 스킵 처리가 되지 않는다.

zip()과 concat()으로 업스트림 결합하기

여러 입력 소스를 하나의 태스크 매핑 iterable로 결합하고 싶은 경우도 흔하다. 파이썬 내장 `zip()`처럼 여러 iterable을 튜플로 묶는 것을 흔히 "zipping"이라 부르며, 이 역시 다운스트림 태스크의 전처리로 수행된다. 이는 태스크 매핑에서 조건부 로직을 쓸 때 특히 유용한데, 예를 들어 S3에서 파일을 내려받으면서 이름을 새로 짓고 싶다면 파일 목록과 새 이름 목록을 `.zip()`으로 묶어 각 다운로드 태스크가 `(원본파일명, 새이름)` 튜플을 받게 만들 수 있다.

파이썬 내장 `zip`처럼, 임의 개수의 iterable을 함께 zip해 위치 인자 개수만큼의 튜플들의 iterable을 얻을 수 있다. 기본적으로 zip된 iterable의 길이는 zip 대상 중 가장 짧은 것과 같아지고, 남는 항목은 버려진다. 선택적 키워드 인자 `default`를 넘기면 파이썬의 `itertools.zip_longest`처럼 동작이 바뀌어, zip된 iterable의 길이가 가장 *긴* 것과 같아지고 부족한 자리는 `default`로 채워진다.

또 다른 흔한 결합 패턴은 여러 iterable에 대해 같은 태스크를 실행하는 것이다 - 물론 각 iterable마다 별도로 같은 코드를 실행해도 되지만(overrider된 `task_id`로 각각 expand), `concat`을 쓰면 Dag가 더 확장성 있고 살펴보기도 쉬워진다. `concat`은 여러 iterable을 순서대로 이어 붙여 하나의 매핑 대상으로 만들며, 임의 개수의 iterable을 함께 concat할 수 있고(`foo.concat(bar, rex)`), 반환값도 XCom 참조이므로 체이닝도 가능하다 (`foo.concat(bar).concat(rex)`) - 파이썬의 `itertools.chain`과 비슷하게 순서를 유지하며 모두 이어 붙인 하나의 iterable이 된다.