Airflow के मुख्य कॉन्सेप्ट्स

Airflow के साथ Data Pipelines बनाना

Volker Janz

Senior Developer Advocate at Astronomer

अपने इंस्ट्रक्टर से मिलें

  वोल्कर यान्ज़ की प्रोफाइल फोटो

$$

Volker Janz

$$

  • सीनियर डेवलपर एडवोकेट, Astronomer
  • गेमिंग में डेटा इंजीनियर के रूप में 14+ साल
  • Airflow के साथ संस्करण 1.x से काम
  • Data Engineer Things में स्पीकर, मेंटर, और न्यूज़लेटर लीड
Airflow के साथ Data Pipelines बनाना

आप क्या बनाएँगे

 

  • TaskFlow API से Dags लिखें
  • टास्क मैपिंग और एसेट-आधारित शेड्यूलिंग से डायनामिक वर्कफ़्लो बनाइए
  • रीट्राई और कॉलबैक से फेल्यर संभालें
  • Airflow के जरिए SQL वर्कलोड चलाएँ

हर चैप्टर की सामग्री का विज़ुअलाइज़ेशन

Airflow के साथ Data Pipelines बनाना

शुरू करने से पहले

 

$$

  • Dags, tasks, और operators में सहज
  • शेड्यूलिंग बेसिक्स से परिचित

Introduction to Airflow - कोर्स पेज

Airflow के साथ Data Pipelines बनाना

त्वरित रिफ्रेशर

from airflow.sdk import dag, task

@dag
def star_wars_dag():


  @task
  def get_star_wars_person():
    import requests
    return requests.get("https://swapi.dev/api/people/1/").json()


  @task.bash
  def print_name(person):
    return f"echo '{person['name']}'"


  person = get_star_wars_person()
  print_name(person)

star_wars_dag()
  • Dag डिपेंडेंसी वाली टास्कों का कलेक्शन है
  • Tasks काम की व्यक्तिगत इकाइयाँ हैं
  • Operators / decorators तय करते हैं हर टास्क क्या करता है
  • Dependencies एग्जिक्यूशन का क्रम तय करती हैं

$$

सिंपल Dag

Airflow के साथ Data Pipelines बनाना

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

$$

Airflow 3 आर्किटेक्चर

 

$$

  • Orchestration: Scheduler, Dag Processor
  • Execution: Worker, Triggerer
  • Interface & Storage: API Server, Metadata DB
Airflow के साथ Data Pipelines बनाना

शेड्यूलिंग अप्रोचेस

# No automatic runs: trigger manually (default)
@dag(schedule=None)
def my_pipeline(): ...


# Time-based: runs every day at 6 AM
@dag(schedule="0 6 * * *")
def daily_pipeline(): ...


# Data-aware: runs when an Asset updates
@dag(schedule=[Asset("my_asset")])
def downstream_pipeline(): ...
Airflow के साथ Data Pipelines बनाना

Dags लिखने के दो तरीके

क्लासिक ऑपरेटर्स

extract = PythonOperator(
    task_id="extract",
    python_callable=extract_fn)
extract >> transform
  • तब चुनें जब डेकोरेटर्स उपलब्ध न हों

TaskFlow API

@task
def extract():
    return {"users": 150}

data = extract()
transform(data)
  • सिंपल Python डेकोरेटर्स
  • कम बॉयलरप्लेट कोड
  • क्लासिक ऑपरेटर्स के साथ मिला सकते हैं
Airflow के साथ Data Pipelines बनाना

इस कोर्स के अभ्यास

$$

IDE अभ्यास स्क्रीनशॉट

 

  • IDE exercises: असली .py फाइलें एडिट करें
  • "Run this file" क्लिक करें या python3 filename.py चलाएँ
  • dag.test() पूरे Dag को एक प्रोसेस में चलाता है
Airflow के साथ Data Pipelines बनाना

अभ्यास करते हैं!

Airflow के साथ Data Pipelines बनाना

Preparing Video For Download...