Emparelhando entradas com zip
Sua equipe tem um pipeline que faz upload de arquivos para locais específicos no armazenamento em nuvem. Um colega usa zip() com expand_kwargs() para emparelhar cada arquivo ao seu destino em nuvem:
@task
def get_files():
return ["users.csv", "orders.csv", "products.csv"]
@task
def get_destinations():
return ["s3://bronze/users/", "s3://bronze/orders/", "s3://bronze/products/"]
@task
def upload(file_to_upload):
# Simulate uploading a file to cloud storage
print(f"Uploading {file_to_upload[0]} to {file_to_upload[1]}")
files = get_files()
destinations = get_destinations()
upload.expand(file_to_upload=files.zip(destinations))
Quantas instâncias de upload o Airflow vai criar?
Este exercicio faz parte do curso
Criando Pipelines de Dados com Airflow
exercicio interativo prático
Transforme teoria em prática com um dos nossos exercicio interativos
Iniciar exercicio