Airflow

expand()의 기본 동작과 파싱 시점 for 루프와의 차이

가장 단순한 형태로는 Dag 파일에 직접 정의된 리스트를, 태스크를 직접 호출하는 대신 `expand()` 함수를 써서 매핑할 수 있다.

이는 파싱 시점에 실행되는 파이썬 for 루프와 다르다 - 동적 태스크 매핑은 태스크 생성을 실행 시점까지 미뤄, 실행 직전에 스케줄러가 업스트림 출력을 보고서야 정확한 태스크 인스턴스 개수를 결정하고 그만큼 복사본(n개)을 만든다. `expand()`에는 키워드 인자만 넘길 수 있다.

매핑된 태스크의 결과를 모아 받는 다운스트림 태스크의 값은 일반 리스트가 아니라 지연 시퀀스(예: `LazySelectSequence([15 items])`)다 - 매핑 개수를 미리 알 수 없기 때문에 필요할 때마다 값을 가져오는 방식으로 동작한다. 인덱싱(`values[0]`)이나 `for` 루프로 순회하는 것은 평범하게 가능하지만, `list(values)`로 변환하면 참조된 모든 업스트림 매핑 태스크의 값을 즉시(eager) 로드하므로 매핑 개수가 크면 성능에 영향을 줄 수 있다. 이 지연 프록시 객체를 그대로 XCom에 push하면 Airflow가 자동으로 list로 강제 변환하면서 성능 저하 가능성을 알리는 경고를 남기며, 태스크 안에서 직접 `list()`를 호출해 명시적으로 변환하면 이 경고를 없앨 수 있다.

매핑된 태스크에 다운스트림(리듀스) 태스크가 없어도, 매핑된 태스크들 자체는 정상적으로 실행된다 - 리듀스 태스크는 필수가 아니다.

태스크가 생성한 리스트로 매핑하기 (Task-generated Mapping)

동적 태스크 매핑의 진짜 힘은 태스크가 순회할 리스트 자체를 생성하게 하는 데서 나온다 - API 호출, DB 조회 등 현재 XCom 백엔드에 저장할 수 있는 것이면 거의 무엇이든 가능하다. `consumer.expand(arg=make_list())`처럼 쓰면 `make_list`가 반환한 리스트의 각 값마다 `consumer`가 한 번씩 호출된다.

다만 업스트림이 이런 태스크 생성 매핑(task-generated mapping)인 경우, 다운스트림에 `trigger_rule=TriggerRule.ALWAYS`를 지정할 수 없다 - `make_list`가 태스크 생성 매핑일 때 확장된 파라미터는 그 태스크가 즉시 실행되는 시점에는 아직 정의되지 않기 때문이다. 이는 태스크와 매핑된 태스크 그룹 모두에 대해 Dag 파싱 시점에 강제되며, 위반하면 에러가 발생한다.

반복 매핑과 고정 인자(partial), 다중 파라미터 cross product

한 매핑 태스크의 결과를 그대로 다음 다운스트림 매핑 태스크의 입력으로 다시 쓸 수도 있다 (반복 매핑, repeated mapping). 예를 들어 `add_one.expand(x=[1, 2, 3])`의 결과를 다시 `add_one.expand(x=first)`로 넘기면 `[3, 4, 5]`가 나온다.

매핑되어 확장되는 인자 외에, 확장마다 바뀌지 않는 고정 인자를 넘기고 싶을 때는 두 함수를 구분해서 쓴다 - 매핑되는 인자는 `expand()`에, 매핑되지 않는 고정 인자는 `partial()`에 넘긴다. 예: `add.partial(y=10).expand(x=[1, 2, 3])`은 `add(x=1, y=10)`, `add(x=2, y=10)`, `add(x=3, y=10)`으로 확장돼 11, 12, 13을 반환한다. 이는 커넥션 ID, 테이블명, 버킷명처럼 확장 전체에 걸쳐 공유하는 값을 넘길 때 특히 유용하다.

하나가 아니라 여러 파라미터를 동시에 `expand`하면, 각 파라미터 값들의 모든 조합(cross product)에 대해 태스크가 생성된다. 예: `add.expand(x=[2, 4, 8], y=[5, 10])`는 `add`를 6번(2와 5, 2와 10, 4와 5, ...) 호출한다 - 다만 이 확장이 실행되는 순서 자체는 보장되지 않는다.