Meer @tasks
Om je workflow verder te bouwen, moet je een extra stap toevoegen om het gedownloade bestand te parsen en de wijzigingen op te slaan. De Dag process_sales is gedefinieerd en heeft de taak pull_file al toegevoegd. In dit geval is de Python-functie al voor je gedefinieerd: parse_file(inputfile, outputfile).
Houd er rekening mee dat je bij het implementeren van Airflow-taken vaak niet precies hoeft te begrijpen wat elke individuele stap doet. Zolang je snapt hoe je de stappen inpast in de structuur van Airflow, kun je de gewenste workflow implementeren.
Deze oefening maakt deel uit van de cursus
Introductie tot Apache Airflow in Python
Oefeninstructies
- Maak een Airflow-taak met de methode
parse_file. - Roep de taak aan met de benodigde argumenten.
Praktische interactieve oefening
Probeer deze oefening eens door deze voorbeeldcode in te vullen.
@dag(dag_id='process_sales')
def process_sales():
# Decorate parse_file as a task
____
def parse_file(inputfile: str, outputfile: str):
with open(inputfile) as infile:
data = json.load(infile)
with open(outputfile, 'w') as outfile:
json.dump(data, outfile)
pull_file('http://dataserver/sales.json', 'latestsales.json')
# Call the parse_file task
____('latestsales.json', 'latestsales_parsed.json')
process_sales()