Mapeamento dinâmico de tarefas

Criando Pipelines de Dados com Airflow

Volker Janz

Senior Developer Advocate at Astronomer

O problema de tarefas fixas no código

@task
def process_file_1():
    transform("/data/file_1.csv")
@task
def process_file_2():
    transform("/data/file_2.csv")
@task
def process_file_3():
    transform("/data/file_3.csv")

Exemplo de mapeamento dinâmico de tarefas

.expand()

@task
def fetch_files():
    return ["/data/file_1.csv", "/data/file_2.csv", "/data/file_3.csv"]

@task(max_active_tis_per_dagrun=2)  # controlar paralelismo máximo
def process_file(path: str):
    print(f"processing file: {path}")


files = fetch_files()
process_file.expand(path=files)

Escalar em tempo de execução

Tarefas mapeadas na UI

Instâncias mapeadas aparecem como instâncias indexadas sob uma única tarefa
Clique em cada uma para ver logs e status individuais
O Dag continua legível mesmo com centenas de instâncias

Tarefas mapeadas na UI

.partial()

@task
def process_file(path: str, output_dir: str):
    transform(path, output_dir)


files = get_files()
process_file.partial(output_dir="/out").expand(path=files)

O problema do produto cartesiano

files = get_files()           # ["/data/a.csv", "/data/b.csv", "/data/c.csv"]
destinations = get_targets()  # ["s3://out/a", "s3://out/b", "s3://out/c"]


# Produto cartesiano: 3 x 3 = 9 instâncias (não é o que queremos!)
process.expand(path=files, dest=destinations)

Unir listas com zip

.zip()

files = get_files()           # ["/data/a.csv", "/data/b.csv", "/data/c.csv"]
destinations = get_targets()  # ["s3://out/a", "s3://out/b", "s3://out/c"]


# Pareado: 3 instâncias (a->a, b->b, c->c)
process.expand_kwargs(files.zip(destinations))

zip() relaciona itens um a um pela posição
expand_kwargs desempacota cada par em argumentos nomeados

Quando usar cada um

Padrão	Caso de uso
`expand()`	Mapear sobre uma única lista
`partial()`	Fixar parâmetros compartilhados entre instâncias
`zip()`	Parear múltiplas listas pela posição

Ajuda a construir Dags escaláveis
Quando os dados mudam, a contagem de tarefas ajusta automaticamente em tempo de execução

Vamos praticar!

Criando Pipelines de Dados com Airflow