Airflow के साथ एक ETL पाइपलाइन बनाना

Apache Airflow का उपयोग करके ETL पाइपलाइनों की दुनिया में आपका स्वागत है। इस ट्यूटोरियल में, हम Polygon API का उपयोग करके स्टॉक मार्केट डेटा खींचने, इस डेटा को रूपांतरित करने, और फिर आसान एक्सेस व हेरफेर के लिए इसे SQLite डेटाबेस में लोड करने पर ध्यान देंगे। आइए शुरू करें!

Apache Airflow और Airflow ETL क्या है?

Apache Airflow डेटा ऑर्केस्ट्रेशन और पाइपलाइन प्रबंधन के लिए उद्योग मानक माना जाता है। जटिल वर्कफ़्लोज़ को ऑर्केस्ट्रेट करने, कार्यों के बीच निर्भरताओं को प्रबंधित करने, विफल कार्यों को पुनः प्रयास करने, और व्यापक लॉगिंग प्रदान करने की क्षमता के कारण यह डेटा वैज्ञानिकों, मशीन लर्निंग इंजीनियरों, और AI प्रैक्टिशनरों में लोकप्रिय हो गया है।

Airflow ETL से तात्पर्य ETL प्रक्रियाओं को प्रबंधित करने के लिए Apache Airflow के उपयोग से है। संक्षेप में, ETL एक प्रकार का डेटा एकीकरण है जिसमें विभिन्न स्रोतों से डेटा निकालना, उसे विश्लेषण के लिए उपयुक्त प्रारूप में रूपांतरित करना, और उसे किसी अंतिम गंतव्य जैसे डेटा वेयरहाउस में लोड करना शामिल होता है। 

हमारे Airflow डेवलपमेंट परिवेश को कॉन्फ़िगर करना 

Airflow के साथ ETL पाइपलाइन बनाना शुरू करने से पहले हमें सबसे पहले अपने डेवलपमेंट परिवेश को कॉन्फ़िगर करना होगा। हमारे डेवलपमेंट परिवेश को कॉन्फ़िगर करने के तरीके के बारे में विस्तृत जानकारी के लिए, हमारा Getting Started with Apache Airflow ट्यूटोरियल देखें।

हमें Astro CLI भी इंस्टॉल करना होगा। कृपया Astronomer देखें, जो Astro CLI को मेंटेन करता है और निर्देशों के लिए विस्तृत दस्तावेज़ प्रदान करता है।

Airflow प्रोजेक्ट बनाना

परिवेश कॉन्फ़िगर करने और Astro CLI इंस्टॉल करने के बाद, हम एक Airflow प्रोजेक्ट बनाते हैं। हम यह टर्मिनल शेल खोलकर और चुने गए पथ पर एक नया डायरेक्टरी बनाकर करते हैं।

 ~/Documents/data-engineering/ETL-pipeline/ संचालित द्वारा

इस डायरेक्टरी की रूट से, आवश्यक संसाधन बनाने के लिए हम निम्न कमांड चलाते हैं:

astro dev initसंचालित द्वारा

डायरेक्टरी की सामग्री कुछ इस प्रकार दिखेगी। सटीक आउटपुट भिन्न हो सकता है।

├── dags/
├── include/
├── plugins/
├── tests/
├── airflow_settings.yaml
├── Dockerfile
├── packages.txt
└── requirements.txtसंचालित द्वारा

अपने प्रोजेक्ट को चालू करने के लिए, निम्न कमांड निष्पादित करें: 

astro dev startसंचालित द्वारा

आपका Airflow परिवेश शुरू होने में लगभग एक मिनट लेगा। जैसे ही यह तैयार हो, अपने वेब ब्राउज़र में localhost: 8080 पर जाएँ, और आपको Airflow UI दिखाई देगा।

अब आप Airflow के साथ अपनी खुद की ETL पाइपलाइन विकसित करना शुरू करने के लिए तैयार हैं!

ETL पाइपलाइन डिज़ाइन करना

कोड लिखना शुरू करने से पहले पाइपलाइन के प्रत्येक घटक की योजना बनाने के लिए समय निकालना महत्वपूर्ण है। खासकर, पहले अपने डेटा स्रोत और उस गंतव्य की पहचान करना अच्छा अभ्यास है जहाँ आपका डेटा लोड होगा। ऐसा करने से हमें यह भी समझ आता है कि रास्ते में डेटा कैसे रूपांतरित किया जाएगा।

हमारे उदाहरण में, हम Polygon API से स्टॉक मार्केट डेटा खींचने, फिर इस डेटा को रूपांतरित करके SQLite डेटाबेस में लोड करने के लिए एक डेटा पाइपलाइन डिज़ाइन करेंगे। इस मामले में, स्रोत सिस्टम Polygon API है, और गंतव्य SQLite डेटाबेस। इसे एक चित्र के साथ दर्शाते हैं: 

स्रोत और गंतव्य आरेख

डेटा इंजीनियर के रूप में हमारे अनुभव से हम जानते हैं कि SQLite डेटाबेस में लोड करने के लिए डेटा तैयार करने हेतु, उसे JSON से सारणीबद्ध प्रारूप में रूपांतरित करना पड़ता है। हम योजना बनाते हैं कि Polygon API से डेटा खींचने के बाद, इसे native Python और pandas लाइब्रेरी का उपयोग करके रूपांतरित किया जाए। आइए इस परिवर्तन को दिखाने के लिए अपने दृश्य को अपडेट करते हैं:

आर्किटेक्चर आरेख

यह अतिरिक्त जानकारी जोड़कर, हमने एक आर्किटेक्चर आरेख बनाया है, जो हमारे सिस्टम का एक व्यापक दृश्य प्रतिनिधित्व है। हम इस पाइपलाइन में तीन लॉजिकल चरण देखते हैं, जो हमारी प्रक्रिया के E, T और L से मेल खाते हैं।

हम इन प्रत्येक कार्यों को एक directed acyclic graph, या DAG, में भी अनुवादित कर सकते हैं, जो एक विशिष्ट कॉन्फ़िगरेशन है जो Airflow द्वारा निष्पादित किए जाने वाले सभी कार्यों के सेट, उनके क्रम, और एक-दूसरे पर उनकी निर्भरताओं को परिभाषित करता है। Directed acyclic graphs के बारे में अधिक जानने के लिए हमारा Introduction to Data Engineering Course लें, जो Airflow DAGs की विस्तार से समीक्षा करता है।

एंटरप्राइज़ डेटा सेटिंग में, कई टीमें तकनीकी विशिष्टताओं (tech specs) नामक अन्य प्रकार के दस्तावेज़ों का उपयोग करती हैं ताकि डिज़ाइन विकल्पों पर सहमति और दस्तावेज़ीकरण किया जा सके। हमारे मामले में, हमारे डेटा पाइपलाइन के लिए डिज़ाइन विकल्पों का दस्तावेज़ीकरण करने हेतु एक तालिका बनाना पर्याप्त है। 

डेटा पाइपलाइन चयन

ध्यान रखें कि हमारा DAG टास्क-स्तर के विवरण पर केंद्रित है और हर जानकारी को कवर नहीं कर सकता। जिन अतिरिक्त विवरणों पर हमें स्पष्टता चाहिए, उन्हें दस्तावेज़ करने के लिए हम दूसरी तालिका का उपयोग कर सकते हैं। हमारे प्रश्नों में शामिल हैं:

• यह DAG कितनी बार चलाया जाएगा? 
• यदि पाइपलाइन में कोई टास्क विफल हो जाए तो क्या होगा? 
• यदि हम अतिरिक्त स्टॉक्स के बारे में डेटा एकत्र करना चाहें तो क्या करें?

सुलझाने के लिए प्रश्न

आइए अब तक की प्रगति का आकलन करें: हमने एक आर्किटेक्चर आरेख बनाया, यह दर्ज किया कि पाइपलाइन को Airflow टास्क्स में कैसे विभाजित किया जा सकता है, और DAG को कॉन्फ़िगर करने के लिए आवश्यक उच्च-स्तरीय जानकारी पहचानी।

यदि आप डेटा पाइपलाइनों के डिज़ाइन, विकास, और परीक्षण के बारे में अधिक सीखना चाहते हैं, तो DataCamp का Introduction to Data Pipelines देखें। यहाँ, आप Python के साथ ETL पाइपलाइनों के निर्माण की बुनियाद सीखेंगे, साथ ही अपने समाधान को मजबूत, लचीला, और पुन: प्रयोज्य बनाने के सर्वोत्तम तरीकों को भी समझेंगे।

Airflow के साथ एक ETL पाइपलाइन बनाना

हम अपने ETL पाइपलाइन के निर्माण को क्रमवार चरणों में व्यवस्थित करेंगे। एक संरचित दृष्टिकोण सुनिश्चित करता है कि प्रत्येक चरण सटीकता के साथ निष्पादित हो। 

Airflow के साथ डेटा निकालना

Polygon API से डेटा खींचने से पहले, हमें Polygon पर जाकर और Create API Key बटन चुनकर एक API टोकन बनाना होगा। कृपया ध्यान दें कि इस ट्यूटोरियल के लिए, पेड subscription की आवश्यकता नहीं है — फ्री टियर एक API key जेनरेट करेगा और हमें आवश्यक सभी फ़ंक्शनलिटी देगा। अपनी API key को कॉपी करके सुरक्षित रखना न भूलें।

API key बनाने के बाद, अब हम Airflow के साथ Polygon API से डेटा निकालना शुरू करने के लिए तैयार हैं। हम अपने बनाए tech specs टेबल का उपयोग करते हैं, जिसमें हमारे DAG कॉन्फ़िगरेशन का विवरण शामिल है, ताकि कोडिंग में मदद मिल सके। 

from airflow import DAG 
from datetime import datetime, timedelta

with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=True,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:संचालित द्वारा

आइए अपना पहला टास्क जोड़ते हैं। हम TaskFlow API और requests मॉड्यूल का उपयोग करके Polygon API से स्टॉक के open-close डेटा को खींचेंगे। 

import requests
...
@task()

def hit_polygon_api(**context):
    # Instantiate a list of tickers that will be pulled and looped over
    stock_ticker = "AMZN"
    # Set variables
    polygon_api_key= "<your-api-key>"
  ds = context.get("ds")

  # Create the URL
    url= f"<https://api.polygon.io/v1/open-close/{stock_ticker}/{ds}?adjusted=true&apiKey={polygon_api_key}>"
    response = requests.get(url)
    # Return the raw data
    return response.json()संचालित द्वारा

इस कोड के बारे में आप कुछ बातें नोटिस करेंगे। 

• hit_polygon_api फ़ंक्शन को @task से सजाया गया है। यह डेकोरेटर फ़ंक्शन को Airflow टास्क में बदल देता है जिसे DAG के हिस्से के रूप में निष्पादित किया जा सकता है।
• hit_polygon_api के सिग्नेचर में context पैरामीटर परिभाषित है। बाद में इसका उपयोग ds कुंजी पर संग्रहीत मान निकालने के लिए किया जाता है।
• context एक डिक्शनरी है जिसमें टास्क और DAG के बारे में मेटाडेटा होता है।
• context डिक्शनरी से ds निकालते समय, हमें data_interaval_end की तारीख YYYY-mm-dd प्रारूप में मिलती है।  
• यह सुनिश्चित करने के लिए कि DAG चलने पर हमारा नया टास्क निष्पादित हो, हमें hit_polygon_api को कॉल करना होगा।

इन सबको मिलाकर, हमारी ETL पाइपलाइन के पहले हिस्से का कोड इस तरह दिखता है।

from airflow import DAG
from airflow.decorators import task
from datetime import datetime, timedelta import requests

with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=False,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:
    # Create a task using the TaskFlow API
    @task()
    def hit_polygon_api(**context):
        # Instantiate a list of tickers that will be pulled and looped over
        stock_ticker = "AMZN"
        # Set variables
        polygon_api_key = "<your-api-key>"
        ds = context.get("ds")
        # Create the URL
        url = f"<https://api.polygon.io/v1/open-close/{stock_ticker}/{ds}?adjusted=true&apiKey={polygon_api_key}>"
        response = requests.get(url)
        # Return the raw data
        return response.json()
    hit_polygon_api()संचालित द्वारा

आप देखेंगे कि 1 January, 2024 के लिए इसे चलाने पर प्रतिक्रिया में एक त्रुटि होगी। चूँकि उस दिन बाजार बंद है, Polygon एक अपवाद का उल्लेख करती हुई प्रतिक्रिया लौटाता है। हम अगले चरण में इसका समाधान करेंगे।

Airflow के साथ डेटा रूपांतरित करना

Polygon API से डेटा निकालने के बाद, हम इसे रूपांतरित करने के लिए तैयार हैं।

इसके लिए, हम TaskFlow API का उपयोग करके एक और टास्क बनाएँगे। इस टास्क का नाम flatten_market_data होगा और इसमें polygon_response पैरामीटर होगा, जो hit_polygon_api फ़ंक्शन द्वारा लौटाया गया कच्चा डेटा है, और **context। हम अभी थोड़ी देर में polygon_response को और करीब से देखेंगे।

वास्तविक रूपांतरण काफ़ी सीधा है। हम Polygon API द्वारा लौटाए गए JSON को एक सूची में समतल करने जा रहे हैं। पेच यह है कि यदि प्रतिक्रिया में कोई कुंजी मौजूद नहीं है, तो हम प्रत्येक कुंजी के लिए अनूठे डिफ़ॉल्ट मान प्रदान करेंगे।

उदाहरण के लिए, यदि प्रतिक्रिया में from कुंजी मौजूद नहीं है, तो हम Airflow के context का उपयोग करके एक डिफ़ॉल्ट मान प्रदान करेंगे। इससे पहले देखी गई समस्या का समाधान होता है, जहाँ प्रतिक्रिया में कुंजियों का सीमित सेट था (क्योंकि बाजार बंद था)। अगला, हम सूची को pandas DataFrame में बदलेंगे और उसे लौटाएँगे। परिणामी टास्क इस प्रकार दिखता है:

@task
def flatten_market_data(polygon_response, **context):
    # Create a list of headers and a list to store the normalized data in
    columns = {
        "status": "closed",
        "from": context.get("ds"),
        "symbol": "AMZN",
        "open": None,
        "high": None,
        "low": None,
        "close": None,
        "volume": None
    }
    # Create a list to append the data to
    flattened_record = []
    for header_name, default_value in columns.items():
        # Append the data
        flattened_record.append(polygon_response.get(header_name, default_value))
    # Convert to a pandas DataFrame
    flattened_dataframe = pd.DataFrame([flattened_record], columns=columns.keys())
    return flattened_dataframeसंचालित द्वारा

हमें hit_polygon_api और flatten_market_data टास्क्स के बीच एक निर्भरता जोड़नी होगी। ऐसा करने के लिए, हम अपने DAG में कोड को नीचे के अनुसार अपडेट करेंगे:

import pandas as pd
with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=True,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:
    @task()
    def hit_polygon_api(**context):
        ...
    @task
    def flatten_market_data(polygon_response, **context):
        # Create a list of headers and a list to store the normalized data in
        columns = {
            "status": None,
            "from": context.get("ds"),
            "symbol": "AMZN",
            "open": None,
            "high": None,
            "low": None,
            "close": None,
            "volume": None
        }
        # Create a list to append the data to
        flattened_record = []
        for header_name, default_value in columns.items():
            # Append the data
            flattened_record.append(polygon_response.get(header_name, default_value))
        # Convert to a pandas DataFrame
        flattened_dataframe = pd.DataFrame([flattened_record], columns=columns.keys())
        return flattened_dataframe
# Set dependencies
    raw_market_data = hit_polygon_api()
    flatten_market_data(raw_market_data)संचालित द्वारा

यहाँ, hit_polygon_api टास्क का रिटर्न मान raw_market_data में संग्रहीत होता है। फिर, raw_market_data को polygon_response पैरामीटर के माध्यम से flatten_market_data टास्क में आर्ग्युमेंट के रूप में पास किया जाता है। यह कोड न केवल hit_polygon_api और flatten_market_data टास्क्स के बीच निर्भरता सेट करता है, बल्कि इन दोनों टास्क्स के बीच डेटा साझा करने की अनुमति भी देता है।

यद्यपि हमने अपेक्षाकृत सरल रूपांतरण किया, Airflow अधिक जटिल डेटा मैनिपुलेशन निष्पादित करने की क्षमता प्रदान करता है। देशी टास्क्स के अलावा, AWS Lambda और DBT जैसे टूल्स का ऑर्केस्ट्रेशन करने के लिए Airflow के व्यापक provider-built hooks और operators का लाभ उठाना आसान है।

Airflow के साथ डेटा लोड करना

हम अपनी ETL पाइपलाइन के अंतिम चरण तक पहुँच गए हैं। हमने इसे SQLite डेटाबेस और TaskFlow API से परिभाषित एक अंतिम टास्क का उपयोग करके करने की योजना बनाई है।

पहले की तरह, टास्क बनाते समय हम एक ही पैरामीटर परिभाषित करेंगे। हम इसका नाम flattened_dataframe रखेंगे। इससे flatten_market_data टास्क द्वारा लौटाए गए डेटा को हमारे नए टास्क में पास किया जा सकेगा।

हमारा डेटा SQLite डेटाबेस में लोड करने के लिए कोड लिखने से पहले, हमें Airflow UI में एक कनेक्शन बनाना होगा। कनेक्शंस पेज खोलने के लिए ये चरण अपनाएँ:

• Airflow UI खोलें
• Admin विकल्प पर होवर करें
• Connections चुनें। 
• नया कनेक्शन बनाने के लिए + आइकन पर क्लिक करें।

आप ऐसे स्क्रीन पर निर्देशित होंगे जो इस तरह दिखता है:

Airflow connections पेज

कनेक्शंस पेज भरने के लिए ये चरण अपनाएँ:

• Connection Type को Sqlite में बदलें।
• Connection Id के लिए "market_database_conn" मान दें।
• Host फ़ील्ड में "/usr/local/airflow/market_data.db" जोड़ें।

इस कनेक्शन की कॉन्फ़िगरेशन नीचे दिए गए चित्र जैसी दिखनी चाहिए। जब ऐसा हो जाए, तो Save पर क्लिक करें।

SQLite डेटाबेस के लिए असहेजा Airflow कनेक्शन

अब जब हमने कनेक्शन बना लिया है, हम SqliteHook का उपयोग करके अपने टास्क में यह जानकारी पुनः प्राप्त कर सकते हैं। नीचे दिए गए कोड पर नज़र डालें।

from airflow.providers.sqlite.hooks.sqlite import SqliteHookसंचालित द्वारा

@task
def load_market_data(flattened_dataframe):
    # Pull the connection
    market_database_hook = SqliteHook("market_database_conn")
    market_database_conn = market_database_hook.get_sqlalchemy_engine()
    # Load the table to Postgres, replace if it exists
    flattened_dataframe.to_sql(
        name="market_data",
        con=market_database_conn,
        if_exists="append",
        index=False
    )
    # print(market_database_hook.get_records("SELECT * FROM market_data;"))संचालित द्वारा

इस कोड के साथ, हम पिछले चरण में निर्दिष्ट SQLite डेटाबेस के लिए एक कनेक्शन बना रहे हैं। फिर, hook से .get_sqlalchemy_engine() मेथड का उपयोग करके कनेक्शन इंजन प्राप्त किया जाता है। इसके बाद इसे flattened_dataframe पर .to_sql() कॉल करते समय con पैरामीटर के आर्ग्युमेंट के रूप में पास किया जाता है।

ध्यान दें कि जिस तालिका में यह डेटा लिखा जा रहा है उसका नाम market_data है, और यदि तालिका मौजूद है, तो उसमें जोड़ दिया जाता है। परीक्षण करते समय, मुझे रिकॉर्ड्स प्राप्त कर उन्हें प्रिंट करके यह जांचना पसंद है कि डेटा लिखा जा रहा है। आप इस टास्क की आखिरी पंक्ति को uncomment करके ऐसा कर सकते हैं।

इन सबको मिलाकर, हमारा कोड कुछ इस प्रकार दिखना चाहिए:

from airflow import DAG
from airflow.decorators import task
from airflow.providers.sqlite.hooks.sqlite import SqliteHook
from datetime import datetime, timedelta
import requests
import pandas as pd

with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=True,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:
    # Create a task using the TaskFlow API
    @task()
    def hit_polygon_api(**context):
        # Instantiate a list of tickers that will be pulled and looped over
        stock_ticker = "AMZN"
        # Set variables
        polygon_api_key = "<your-api-key>"
        ds = context.get("ds")
        # Create the URL
        url = f"<https://api.polygon.io/v1/open-close/{stock_ticker}/{ds}?adjusted=true&apiKey={polygon_api_key}>"
        response = requests.get(url)
        # Return the raw data
        return response.json()
    @task
    def flatten_market_data(polygon_response, **context):
        # Create a list of headers and a list to store the normalized data in
        columns = {
            "status": None,
            "from": context.get("ds"),
            "symbol": "AMZN",
            "open": None,
            "high": None,
            "low": None,
            "close": None,
            "volume": None
        }
        # Create a list to append the data to
        flattened_record = []
        for header_name, default_value in columns.items():
            # Append the data
            flattened_record.append(polygon_response.get(header_name, default_value))
        # Convert to a pandas DataFrame
        flattened_dataframe = pd.DataFrame([flattened_record], columns=columns.keys())
        return flattened_dataframe
    @task
    def load_market_data(flattened_dataframe):
        # Pull the connection
        market_database_hook = SqliteHook("market_database_conn")
        market_database_conn = market_database_hook.get_sqlalchemy_engine()
        # Load the table to SQLite, append if it exists
        flattened_dataframe.to_sql(
            name="market_data",
            con=market_database_conn,
            if_exists="append",
            index=False
        )
    # Set dependencies between tasks
    raw_market_data = hit_polygon_api()
    transformed_market_data = flatten_market_data(raw_market_data)
    load_market_data(transformed_market_data)संचालित द्वारा

फिर से, हमने अपनी निर्भरताएँ अपडेट की हैं ताकि flatten_market_data टास्क से लौटाया गया डेटा load_market_data टास्क को पास किया जा सके। हमारे DAG के लिए परिणामी ग्राफ़ व्यू कुछ इस प्रकार दिखता है:

ETL पाइपलाइन के लिए ग्राफ़ व्यू

परीक्षण

अब जब आपने अपना पहला Airflow DAG बना लिया है, तो यह सुनिश्चित करने का समय है कि यह काम करता है। ऐसा करने के कई तरीके हैं, लेकिन सबसे सामान्य में से एक DAG को एंड-टू-एंड चलाना है।

Airflow UI में ऐसा करने के लिए, आप अपने DAG पर जाएँ और स्विच को नीले से active पर टॉगल करें। चूँकि catchup को True पर सेट किया गया था, एक DAG रन कतारबद्ध होना चाहिए और चलना शुरू होना चाहिए। यदि कोई टास्क सफलतापूर्वक निष्पादित होता है, तो UI में उससे संबंधित बॉक्स हरा हो जाएगा। यदि DAG के सभी टास्क सफल होते हैं, तो DAG को success के रूप में चिह्नित किया जाएगा, और अगला DAG रन ट्रिगर होगा।

यदि कोई टास्क विफल होता है, तो स्थिति up for retry होगी और पीले रंग में चिह्नित होगी। जब ऐसा होता है, तो उस टास्क के लॉग्स देखना सबसे अच्छा होता है। यह grid view में पीले बॉक्स पर क्लिक करके और Logs चुनकर किया जा सकता है। यहाँ, आपको exception संदेश मिलेगा और आप ट्रायेज़ शुरू कर सकते हैं। यदि कोई टास्क निर्दिष्ट retries की संख्या से अधिक बार विफल होता है, तो उस टास्क और DAG की स्थिति failed पर सेट कर दी जाएगी।

DAG को एंड-टू-एंड परीक्षण करने के अतिरिक्त, Airflow यूनिट टेस्ट लिखना आसान बनाता है। जब आप प्रारंभ में अपना परिवेश astro dev start का उपयोग करके बनाते हैं, तो आपके लिए एक tests/ डायरेक्टरी बनाई जाती है। यहाँ, आप अपने DAG के साथ-साथ DAG के घटकों के लिए भी यूनिट टेस्ट जोड़ सकते हैं।

नीचे हमारे DAG के कॉन्फ़िगरेशन के लिए एक यूनिट टेस्ट है। यह टेस्ट उन कई पैरामीटरों को मान्य करता है जो define करते समय सेट किए गए थे, जैसे start_date, schedule, और catchup। अपना टेस्ट लिखने के बाद, अपने प्रोजेक्ट की रूट पर जाएँ, और चलाएँ:

from airflow.models.dagbag import DagBag
from datetime import datetime
import pytz

def test_market_etl_config():
    # Pull the DAG
    market_etl_dag = DagBag().get_dag("market_etl")
    # Assert start date, schedule, and catchup
    assert market_etl_dag.start_date == datetime(2024, 3, 25, 9, tzinfo=pytz.UTC)
    assert market_etl_dag.schedule_interval == "@daily"
    assert market_etl_dag.catchupसंचालित द्वारा

astro dev pytestसंचालित द्वारा

यह कमांड आपकी tests/ डायरेक्टरी के सभी यूनिट टेस्ट्स को निष्पादित करेगा। यदि आप केवल एक ही टेस्ट निष्पादित करना चाहते हैं, तो उपरोक्त कमांड के आर्ग्युमेंट के रूप में फ़ाइल का पथ प्रदान कर सकते हैं। Astro CLI का उपयोग करके टेस्ट निष्पादित करने के अलावा, कोई भी Python टेस्ट रनर यूनिट टेस्ट लिखने और निष्पादित करने के लिए उपयोग किया जा सकता है।

व्यक्तिगत प्रोजेक्ट्स के लिए, यूनिट टेस्ट लिखना यह सुनिश्चित करने में मदद करेगा कि आपका कोड अपेक्षा के अनुरूप प्रदर्शन करता है। एंटरप्राइज़ सेटिंग में, यूनिट टेस्ट लगभग हमेशा आवश्यक होते हैं। अधिकांश डेटा टीमें अपने Airflow प्रोजेक्ट को डिप्लॉय करने के लिए किसी न किसी प्रकार के CI/CD टूल का लाभ उठाएँगी। इस प्रक्रिया में आम तौर पर यूनिट टेस्ट चलाना और उनके आउटपुट को मान्य करना शामिल होता है ताकि यह सुनिश्चित हो सके कि आपने जो DAG लिखा है वह प्रोडक्शन-रेडी है। यूनिट टेस्ट्स के बारे में अधिक जानकारी के लिए, हमारा How to Use Pytest for Unit Testing ट्यूटोरियल और Introduction to Testing in Python देखें।

उन्नत Airflow टिप्स और तकनीकें

हमने एक सरल डेटा पाइपलाइन बनाई है जो काम करती है, और हमने रूपांतरण और स्थायित्व तकनीकों का भी उपयोग किया है। अन्य मामलों में, Airflow provider-built और कस्टम ऑपरेटरों का उपयोग करके जटिल वर्कफ़्लोज़ को ऑर्केस्ट्रेट करने में सक्षम है, जो टेराबाइट्स डेटा प्रोसेस कर सकता है।

इनमें से कुछ उदाहरण हैं S3ToSnowflakeOperator और DatabricksRunNowOperator, जो बड़े डेटा स्टैक के साथ एकीकरण को आसान बनाते हैं। ऐसे ऑपरेटरों के साथ शौकिया-सेटिंग में काम करना मुश्किल हो सकता है। उदाहरण के लिए, S3ToSnowflakeOperator का उपयोग करने के लिए, आपको AWS और Snowflake दोनों खातों और उन संसाधनों की कॉन्फ़िगरेशन की आवश्यकता होगी जिनके बीच आप डेटा ट्रांसफर करेंगे।

ETL वर्कफ़्लोज़ के अलावा, Airflow ELT वर्कफ़्लोज़ का भी समर्थन करता है, जो क्लाउड डेटा वेयरहाउस का लाभ उठाने वाली टीमों के लिए तेजी से उद्योग मानक बनता जा रहा है। डेटा पाइपलाइनों को डिज़ाइन करते समय इसे ध्यान में रखें।

हमारी पाइपलाइन के load घटक में, हमने एक SQLite डेटाबेस के लिए एक कनेक्शन बनाया, जिसे बाद में पुनः प्राप्त किया गया और डेटा को स्थायी बनाने के लिए उपयोग किया गया। Connections, जिन्हें कभी-कभी Secrets भी कहा जाता है, Airflow में एक फीचर है जिसे स्रोत और गंतव्य सिस्टम के साथ इंटरैक्शन को सरल बनाने के लिए डिज़ाइन किया गया है। इन कनेक्शंस का उपयोग करके संवेदनशील जानकारी, जैसे आपकी Polygon API key, को स्टोर करने से, आपके कोड की सुरक्षा बढ़ती है। यह तरीका आपको अपने कोडबेस से अलग क्रेडेंशियल्स प्रबंधित करने की भी अनुमति देता है। जहाँ भी संभव हो, अपने वर्कफ़्लो को सुरक्षित और व्यवस्थित रखने के लिए Connections का व्यापक रूप से उपयोग करना उचित है।

आपने देखा होगा कि “AMZN” स्टॉक टिकर को हमारे hit_polygon_api और flatten_market_data टास्क्स में हार्ड-कोड किया गया था। इससे हमें इसी एक स्टॉक टिकर के लिए डेटा निकालने, रूपांतरित करने, और लोड करने की अनुमति मिली। लेकिन यदि आप इसी कोड को कई स्टॉक टिकर के लिए उपयोग करना चाहें तो? सौभाग्य से, DAGs को डायनेमिक रूप से जेनरेट करना आसान है। हमारे कोड में मामूली रीफ़ैक्टरिंग के साथ, हम स्टॉक टिकर्स की एक सूची पर लूप चला सकते हैं और टिकर मानों को पैरामीटराइज़ कर सकते हैं। इससे आपके DAGs अधिक मॉड्यूलर और पोर्टेबल बनते हैं। डायनेमिक रूप से DAGs जेनरेट करने के बारे में जानकारी के लिए, Astronomer का Dynamically Generate DAGs in Airflow दस्तावेज़ देखें।

निष्कर्ष

बधाई हो! आपने Python, pandas, और SQLite का उपयोग करके Polygon API से स्टॉक मार्केट डेटा निकालने, रूपांतरित करने, और लोड करने के लिए एक Airflow DAG बनाया है। इस दौरान, आपने आर्किटेक्चर आरेखों और टेक स्पेक्स बनाने, Airflow कनेक्शंस सेट करने, और अपने DAGs का परीक्षण करने में अपनी कौशल को निखारा। जैसे-जैसे आप Airflow की यात्रा जारी रखते हैं, अधिक उन्नत तकनीकों के साथ प्रयोग करें ताकि आपकी पाइपलाइंस मजबूत, लचीली, और पुन: प्रयोज्य बन सकें।  

और अधिक जानने के लिए कुछ अतिरिक्त संसाधन देखें: 

• Introduction to Airflow in Python कोर्स
• Getting Started with Apache Airflow ट्यूटोरियल
• Airflow बनाम Prefect तुलना
• Airflow के शीर्ष विकल्प
• Airflow के शीर्ष इंटरव्यू प्रश्न