У тебя даг должен быть идемпотентным (сколько раз ни запускай – результат будет один), а само определение workflow должно быть статичным (в любой момент можно ответить на вопрос "сколько тасков, и как они связаны между собой").
Т.е. логично в рантайме делать, например, так:
1 таск – ShortCircuitOperator, например. В нём получить список того, что нужно будет обработать, положить в Variable/XCOM
2 таск – достать и обработать.
Если нужно распараллелить для скорости, например, отгрузили 200 новых файлов, а хочется их обрабатывать не по одному, а на куче воркеров, то можно класть в JSON-е словарь в Variable, а уже последующим таском доставать соответствующие значения (task_i берёт значения из ключа i).
