Mehr @tasks
Um mit der Implementierung deines Workflows fortzufahren, musst du einen weiteren Schritt hinzufügen, um die Änderungen der heruntergeladenen Datei zu parsen und zu speichern. Der Dag process_sales ist definiert und hat den Task pull_file bereits hinzugefügt. In diesem Fall ist die Python-Funktion bereits für dich definiert: parse_file(inputfile, outputfile).
Beachte, dass du beim Implementieren von Airflow-Tasks die einzelnen Schritte, die dir gegeben werden, nicht unbedingt im Detail verstehen musst. Solange du weißt, wie du die Schritte in die Struktur von Airflow einbettest, kannst du den gewünschten Workflow umsetzen.
Diese Übung ist Teil des Kurses
<Kurs>Einführung in Apache Airflow mit Python</Kurs>Übungsanweisungen
- Erstelle einen Airflow-Task mit der Methode
parse_file. - Rufe den Task mit den notwendigen Argumenten auf.
Interaktive praktische Übung
Versuche dich an dieser Übung, indem du diesen Beispielcode vervollständigst.
@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()