और @tasks
अपने वर्कफ़्लो को आगे बढ़ाने के लिए, आपको डाउनलोड की गई फ़ाइल के बदलावों को पार्स करके सेव करने का एक और स्टेप जोड़ना है। Dag process_sales परिभाषित है और उसमें pull_file टास्क पहले से जोड़ा गया है। इस केस में, Python फंक्शन आपके लिए पहले से परिभाषित है: parse_file(inputfile, outputfile).
ध्यान दें कि Airflow टास्क लागू करते समय, अक्सर आपको दिए गए हर स्टेप का पूरा मतलब समझ में आए, यह ज़रूरी नहीं होता। जब तक आप समझते हैं कि इन स्टेप्स को Airflow की संरचना में कैसे लपेटना है, आप मनचाहा वर्कफ़्लो लागू कर पाएँगे.
यह अभ्यास पाठ्यक्रम का हिस्सा है
Python में Apache Airflow परिचय
अभ्यास निर्देश
parse_fileमेथड का उपयोग करके एक Airflow टास्क बनाएँ.- टास्क को ज़रूरी आर्ग्युमेंट्स के साथ कॉल करें.
इंटरैक्टिव व्यावहारिक अभ्यास
इस अभ्यास को इस नमूना कोड को पूरा करके आज़माएँ।
@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()