Associer des entrées avec zip
Votre équipe dispose d'un pipeline qui envoie des fichiers vers des emplacements précis de stockage cloud. Un collègue utilise zip() avec expand_kwargs() pour associer chaque fichier à sa destination de stockage :
@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))
Combien d'instances de upload Airflow va-t-il créer ?
Cet exercice fait partie du cours
<cours>Créer des pipelines de données avec Airflow</cours>Exercice interactif pratique
Transformez la théorie en action avec l’un de nos exercices interactifs
Commencer l’exercice