Daha fazla @tasks
İş akışını uygulamaya devam etmek için, indirilen dosyanın değişikliklerini ayrıştırıp kaydeden bir adım daha eklemen gerekiyor. process_sales adlı Dag tanımlandı ve pull_file görevi zaten eklendi. Bu durumda, Python fonksiyonu senin için hazır: parse_file(inputfile, outputfile).
Unutma, Airflow görevlerini uygularken çoğu zaman sana verilen adımların her birini tam olarak anlamayabilirsin. Adımları Airflow'un yapısı içinde nasıl sarmalayacağını bildiğin sürece, hedeflenen iş akışını hayata geçirebilirsin.
Bu egzersiz, kursun bir parçasıdır
Python ile Apache Airflow’a Giriş
Egzersiz talimatları
parse_filemetodunu kullanarak bir Airflow görevi oluştur.- Görevi gerekli argümanlarla çağır.
Uygulamalı etkileşimli egzersiz
Bu egzersizi bu örnek kodu tamamlayarak deneyin.
@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()