Airflow का उपयोग कैसे करें: विश्वसनीय डेटा पाइपलाइन बनाने के लिए एक व्यावहारिक, एंड-टू-एंड गाइड
यदि आप डेटा स्थानांतरित करते हैं या ML कार्यों को व्यवस्थित करते हैं, तो आपने शायद यही बात सुनी होगी: “बस इसे Airflow में डाल दें।” सच्चाई यह है कि Apache Airflow तब चमकता है जब आपको जटिल वर्कफ़्लो पर दृश्यता, विश्वसनीयता और नियंत्रण की आवश्यकता होती है। इस व्यावहारिक गाइड में, हम Airflow का उपयोग करने के तरीके के बारे में चरण-दर-चरण जानेंगे—मूल अवधारणाओं से लेकर उत्पादन-तैयार पैटर्न तक—ताकि आप उन पाइपलाइनों को भेज सकें जिन पर आपको भरोसा है।
हम इसे क्रियाशील रखेंगे: आपको DAGs और कार्यों के लिए एक मानसिक मॉडल, TaskFlow API के साथ हैंड्स-ऑन उदाहरण, परिनियोजन विकल्प, परीक्षण रणनीतियाँ और सर्वोत्तम अभ्यास मिलेंगे। अंत तक, आप "मैं ट्यूटोरियल चला सकता हूँ" से "मैं इसे prod में चला सकता हूँ" तक पहुँच जाएँगे।
ध्यान दें: गहन जानकारी और संदर्भ के लिए, आधिकारिक दस्तावेज़ उत्कृष्ट हैं और नियमित रूप से अपडेट किए जाते हैं।
Apache Airflow वास्तव में क्या है?
Airflow एक ऑर्केस्ट्रेटर है—डेटा प्रोसेसर नहीं। यह आपके द्वारा कहीं और चलाए जा रहे कार्यों (डेटाबेस, वेयरहाउस, Spark कार्य, API, कंटेनर) को शेड्यूल, ऑर्डर और मॉनिटर करता है। आप वर्कफ़्लो को DAGs (डायरेक्टेड एसाइक्लिक ग्राफ़) के रूप में परिभाषित करते हैं, जो केवल पायथन फ़ाइलें हैं जो कार्यों और उनकी निर्भरताओं को एन्कोड करती हैं। Airflow तब आपके शेड्यूल, मापदंडों और वातावरण के अनुसार उन कार्यों को निष्पादित करता है।
- DAG: वर्कफ़्लो परिभाषा (निर्भरताओं वाले कार्यों का ग्राफ़)।
- कार्य: कार्य की एक इकाई (पायथन फ़ंक्शन, SQL निष्पादन, Bash कमांड, बाहरी कार्य ट्रिगर, आदि)।
- ऑपरेटर: एक प्रकार के कार्य के लिए एक टेम्पलेट (उदाहरण के लिए,
PythonOperator, BashOperator, KubernetesPodOperator)।
- शेड्यूलर: यह तय करता है कि क्या चलाना है और कब चलाना है।
- एक्जीक्यूटर: कार्यों को चलाता है (स्थानीय रूप से, Celery, Kubernetes, आदि के साथ)।
- UI: रन, लॉग, पुनर्प्रयास और वंश के लिए आपका नियंत्रण केंद्र।
Airflow स्थापित करने के बाद आधिकारिक ट्यूटोरियल से शुरुआत करें; वे आपको जल्दी से बड़ी तस्वीर देते हैं।
Airflow को सही तरीके से स्थापित और चलाना
Airflow लचीला है। वह पथ चुनें जो आपके चरण से मेल खाता हो:
- स्थानीय विकास (त्वरित शुरुआत):
- परियोजना द्वारा प्रदान किए गए त्वरित-शुरुआत Docker Compose का उपयोग करें। यह वेबसर्वर, शेड्यूलर, डेटाबेस और अधिक को समझदार डिफ़ॉल्ट के साथ स्पिन करता है।
- DAGs सीखने और पुनरावृति करने के लिए बढ़िया।
- प्रबंधित Postgres के साथ Celery Executor या Kubernetes Executor।
- S3/GCS में लॉग स्टोर करें और अपनी छवि या
requirements.txt के साथ निर्भरताओं को पैकेज करें।
- लोच के लिए Kubernetes Executor या ऑटोस्केलिंग वर्कर्स के साथ Celery Executor।
- अपग्रेड के लिए बाहरी रहस्य (Vault), मजबूत अवलोकन क्षमता (लॉग + मेट्रिक्स), और ब्लू/ग्रीन परिनियोजन।
टिप: पदोन्नति से पहले अपने Airflow कोडबेस को संस्करण-नियंत्रित, कंटेनरीकृत और परीक्षणित रखें। "सर्वोत्तम अभ्यास" पृष्ठ उत्पादन-तैयार पैटर्न की रूपरेखा बताता है।
मुख्य अवधारणाएँ जिनका आप प्रतिदिन उपयोग करेंगे
DAGs: कोड के रूप में आपका वर्कफ़्लो
एक DAG एक पायथन फ़ाइल है जो परिभाषित करती है:
- DAG मेटाडेटा: आईडी, शेड्यूल, प्रारंभ तिथि, टैग।
- डिफ़ॉल्ट तर्क: पुनर्प्रयास, मालिक, SLA।
- कार्य और उनकी निर्भरताएँ।
एक DAG को "क्या" और "कब" के रूप में सोचें, और कार्यों को "कैसे" के रूप में।
कार्य और ऑपरेटर
ऑपरेटर सामान्य कार्यों के लिए प्रीफैब हैं। उदाहरण:
- पायथन कोड के लिए PythonOperator / TaskFlow
@task
- शेल कमांड के लिए BashOperator
- API के लिए SimpleHttpOperator
- कंटेनरीकृत कार्यों के लिए KubernetesPodOperator
- वेयरहाउस कार्य के लिए SQL प्रदाता (जैसे, Snowflake, BigQuery, Postgres)
TaskFlow API: आधुनिक, पायथोनिक तरीका
TaskFlow API आपको @task के साथ पायथन फ़ंक्शन के रूप में कार्य लिखने, XCom के माध्यम से पास होने वाले मानों को वापस करने और उन्हें साफ-सुथरा बनाने की अनुमति देता है। यह बॉयलरप्लेट को कम करता है और पठनीयता में सुधार करता है—अत्यधिक अनुशंसित।
आपका पहला Airflow DAG (TaskFlow संस्करण)
नीचे एक न्यूनतम ETL-शैली का उदाहरण दिया गया है जो प्रमुख विचारों को दर्शाता है: शेड्यूलिंग, TaskFlow, निर्भरताएँ और XCom डेटा पासिंग।
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
with DAG(
dag_id="weather_etl_example",
start_date=datetime(2024, 1, 1),
schedule="0 * * * *", # hourly
catchup=False,
tags=.
शेड्यूलिंग, कैचअप और बैकफिल
schedule: Cron या प्रीसेट (@daily, @hourly)।
start_date + catchup: यदि catchup=True है, तो Airflow प्रारंभ तिथि से रन को बैकफिल करेगा। स्ट्रीमिंग-शैली पाइपलाइन के लिए, catchup=False सेट करें।
- मैनुअल बैकफिल: ऐतिहासिक अंतराल को फिर से चलाने के लिए UI या CLI का उपयोग करें।
अंगूठे का व्यावहारिक नियम: नियतात्मक बैच कार्यों के लिए कैचअप सक्षम करें; रीयल-टाइम या API दर-सीमित पाइपलाइनों के लिए अक्षम करें।
कार्यों के बीच डेटा (XCom) को सुरक्षित रूप से पास करना
- छोटी वस्तुएँ: TaskFlow के साथ मान लौटाना ठीक है।
- बड़े पेलोड: XCom में एक कुंजी के साथ ऑब्जेक्ट स्टोरेज (S3/GCS) में स्टोर करें।
- XCom में संवेदनशील डेटा से बचें; रहस्य बैकएंड (जैसे, Vault) और पर्यावरण चर का उपयोग करें।
डायनामिक टास्क मैपिंग और फैन-आउट वर्कलोड
Airflow इनपुट के आधार पर रनटाइम पर कार्यों को गतिशील रूप से उत्पन्न कर सकता है—विभाजित डेटासेट या बहु-किरायेदार कार्यों के लिए आदर्श।
- DAGs को नियतात्मक और आइडेंम्पोटेंट रखें।
- ऑर्केस्ट्रेशन (Airflow) को संगणना (Spark, dbt, वेयरहाउस) से अलग करें।
- स्पष्टता और XCom स्वच्छता के लिए TaskFlow API का उपयोग करें।
- DAGs को पैरामीटराइज़ करें; चरों का समझदारी से उपयोग करें।
- अपनी पाइपलाइनों की निगरानी, अलर्ट और दस्तावेज़ बनाएँ।
डेटा वेयरहाउस और ML के साथ कैसे काम करें
- डेटा वेयरहाउस: SQL कार्यों के लिए प्रदाता ऑपरेटरों (जैसे, SnowflakeOperator, BigQueryInsertJobOperator) का उपयोग करें। SQL को फ़ाइलों या संस्करणित मॉड्यूल में स्टोर करें।
- dbt: Bash/KubernetesPodOperator या प्रदाताओं में समर्पित dbt ऑपरेटरों के माध्यम से dbt को ट्रिगर करें।
- ML: सुविधा पीढ़ी, प्रशिक्षण और बैच अनुमान को अलग-अलग कार्यों के रूप में व्यवस्थित करें; स्टोरेज में आर्टिफैक्ट को कैश करें और मेट्रिक्स को लॉग करें।
उन्नत शेड्यूलिंग: डेटासेट और क्रॉस-DAG निर्भरताएँ
- डेटासेट एक DAG को एक तार्किक डेटासेट का उत्पादन करने देते हैं जो अपडेट होने पर दूसरे DAG को ट्रिगर करता है—तदर्थ ट्रिगर की तुलना में क्लीनर।
- विरासत पैटर्न के लिए, ExternalTaskSensor काम करता है, लेकिन डेटासेट अधिक घोषणात्मक होते हैं।
सुरक्षा और अनुपालन
- UI में भूमिका-आधारित एक्सेस कंट्रोल (RBAC) का उपयोग करें।
- टीम या विश्वास सीमा के अनुसार वातावरण को अलग करें।
- लॉग और कनेक्शन परिवर्तन इतिहास के माध्यम से ऑडिट ट्रेल्स रखें।
अपग्रेड और संस्करण
- उत्पादन-जैसे वर्कलोड के साथ स्टेजिंग में अपग्रेड का परीक्षण करें।
- जानबूझकर प्रदाताओं को पिन और अपग्रेड करें।
- एक्जीक्यूटर-विशिष्ट परिवर्तनों और अप्रचलन के लिए रिलीज़ नोट्स पढ़ें।
आपके पहले उत्पादन DAG के लिए एक त्वरित चेकलिस्ट
- स्पष्ट स्वामित्व (
owner टैग) और अलर्ट कॉन्फ़िगर किए गए।
- उचित बैकऑफ़ के साथ
retries सेट करें।
- आइडेंम्पोटेंट कार्य और स्पष्ट निर्भरताएँ।
- छोटे XCom पेलोड; स्टोरेज में बड़ा डेटा।
- लॉग टिकाऊ स्टोरेज में भेजे गए; मेट्रिक्स निर्यात किए गए।
- रोलआउट योजना (कैनरी या ब्लू/ग्रीन) और रोलबैक चरण।
उदाहरण: एक यथार्थवादी वेयरहाउस लोड DAG
यह पैटर्न दैनिक फ़ाइलों को निकालता है, उन्हें मान्य करता है, और उन्हें विभाजन के अनुसार गतिशील मैपिंग और डिफरएबल सेंसर के साथ एक वेयरहाउस टेबल में लोड करता है।
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
from airflow.sensors.filesystem import FileSensor
- उत्पादन में प्रचार करने से पहले सर्वोत्तम अभ्यास की समीक्षा करें।
- अपनी प्रणालियों (वेयरहाउस, क्लाउड, ML टूल) के लिए प्रदाता दस्तावेज़ों का अन्वेषण करें।
वैसे: AI साइडकिक के साथ लेखन को गति दें
ध्यान देने योग्य: यदि आप बहुत सारे DAGs का मसौदा तैयार करते हैं, तो एक AI सहायक जो कोड को समझता है, बॉयलरप्लेट को तेज कर सकता है, TaskFlow स्टब्स उत्पन्न कर सकता है और यहां तक कि निर्भरता सुधार का सुझाव भी दे सकता है। यदि आप अपने संपादक और ब्राउज़र के साथ एक हल्का सहायक चाहते हैं, तो Sider.AI विकास के दौरान त्वरित कोड पुनर्लेखन और स्पष्टीकरण के लिए उपयोगी हो सकता है। मुख्य बातें
- Airflow का उपयोग ऑर्केस्ट्रेट करने के लिए करें, कंप्यूट करने के लिए नहीं।
- साफ, परीक्षण योग्य DAGs के लिए TaskFlow API को प्राथमिकता दें।
- डेटा को XCom से बाहर रखें; इसके बजाय संदर्भ पास करें।
- स्लॉट बचाने के लिए डिफरएबल सेंसर/ऑपरेटरों का उपयोग करें।
- कंटेनराइज़ करें, परीक्षण करें और वातावरण के माध्यम से प्रचार करें।
- अपने उत्तरी तारे के रूप में आधिकारिक ट्यूटोरियल और सर्वोत्तम प्रथाओं पर भरोसा करें।
FAQ
Q1:Airflow का उपयोग करना सीखने का सबसे आसान तरीका क्या है?
DAGs, कार्यों, शेड्यूलिंग और UI को समझने के लिए आधिकारिक ट्यूटोरियल से शुरुआत करें। फिर एक छोटी TaskFlow-आधारित पाइपलाइन बनाएँ और उत्पादन-तत्परता के लिए सर्वोत्तम अभ्यास गाइड के साथ पुनरावृति करें।
Q2:मुझे Airflow में TaskFlow API या क्लासिक ऑपरेटरों का उपयोग करना चाहिए?
अधिकांश पायथोनिक पाइपलाइनों के लिए TaskFlow API का उपयोग करें क्योंकि यह क्लीनर है और XCom रिटर्न को स्वाभाविक रूप से संभालता है। क्लासिक ऑपरेटर अभी भी गैर-पायथन कार्यों जैसे Bash, SQL या कंटेनर कार्यों के लिए बहुत अच्छे हैं।
Q3:मैं Airflow कार्यों के बीच बड़े डेटा को कैसे पास करूँ?
XCom में बड़े पेलोड डालने से बचें। डेटा को S3/GCS या डेटाबेस में स्टोर करें और कार्यों को तेज़ और विश्वसनीय रखने के लिए केवल संदर्भ या URIs को XCom के माध्यम से पास करें।
Q4:मुझे उत्पादन में Airflow के लिए कौन सा एक्जीक्यूटर चुनना चाहिए?
लोच और अलगाव के लिए, Kubernetes Executor एक मजबूत डिफ़ॉल्ट है। सरल सेटअप के लिए, Celery Executor अच्छी तरह से काम करता है—बस ऑटोस्केलिंग, मजबूत लॉगिंग और बाहरीकृत रहस्यों को सुनिश्चित करें।
Q5:मैं कई Airflow DAGs में निर्भरताओं को कैसे संभालूँ?
घोषणात्मक क्रॉस-DAG ट्रिगर के लिए डेटासेट का उपयोग करें जब एक पाइपलाइन दूसरे के लिए डेटा का उत्पादन करती है। वैकल्पिक रूप से, ExternalTaskSensor रन को समन्वयित कर सकता है, लेकिन डेटा-संचालित ऑर्केस्ट्रेशन के लिए डेटासेट क्लीनर होते हैं।
