Inputs koppelen met zip
Je team heeft een pipeline die bestanden uploadt naar specifieke cloudopslaglocaties. Een collega gebruikt zip() met expand_kwargs() om elk bestand te koppelen aan zijn cloudopslagbestemming:
@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):
# Simuleer het uploaden van een bestand naar cloudopslag
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))
Hoeveel instanties van upload maakt Airflow aan?
Deze oefening maakt deel uit van de cursus
Data-pijplijnen bouwen met Airflow
Interactieve oefening met praktijkervaring
Zet theorie om in actie met een van onze interactieve oefeningen
Begin oefening