Data Collect Kar
Rahe Ho — Par
Pahunch Raha Hai?

Cron Jobs Ka Problem — Raat Ko 2 Baje Pata Chalta Hai

Bahut saari teams ka yahi haal hota hai. Ek pipeline banana tha — SQL se data nikalo, Python se clean karo, final table mein load karo. Solution? Cron job lagao, scripts schedule karo. Done.

Lekin production mein yeh approach silently toot jaati hai. Ek script fail hoti hai — koi nahi jaanta. Dependency break ho jaata hai — data half-processed rehta hai. System restart hota hai — cron job dobaara start nahi hota. Subah office aate ho toh dashboard empty milta hai.

Cron job ek scheduler hai, orchestrator nahi. Yeh sirf ek script ko ek time pe chalata hai. Isse na pata hai ki pehli script complete hui ya fail hui. Na yeh retry karega. Na yeh alert bhejega. Na yeh dependencies track karega.

Data Pipeline Kya Hoti Hai?

Data Pipeline ek automated workflow hai jo data ko ek source se uthata hai, use process karta hai, aur ek defined destination pe load karta hai. Har step ek specific kaam karta hai — aur yeh sab ek defined order mein hota hai.

DAG — Directed Acyclic Graph Kya Hota Hai?

Jab pipeline mein multiple steps hote hain — aur kuch steps parallel mein chal sakte hain, kuch sequential hain — tab hum DAG use karte hain define karne ke liye ki konsa task kab chalega.

Directed — har task ka ek direction hai. Task A se Task B — backward nahi jaata.
Acyclic — koi cycle nahi. Task A → Task B → Task A nahi ho sakta — infinite loop nahi hoga.
Graph — tasks nodes hain, dependencies edges hain.

                    ┌─────────────────────┐
                    │   START (Trigger)   │
                    │   Daily @ 2:00 AM   │
                    └──────────┬──────────┘
                               │
                               ▼
                    ┌─────────────────────┐
                    │  extract_raw_sales  │  ← Task A: S3 se raw CSV pull karo
                    └──────────┬──────────┘
                               │
               ┌───────────────┼───────────────┐
               ▼               ▼               ▼
      ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
      │ clean_nulls  │ │ remove_dupes │ │ fix_formats  │  ← Parallel tasks
      └──────┬───────┘ └──────┬───────┘ └──────┬───────┘
             └───────────────┬┘               ─┘
                             │
                             ▼
                  ┌─────────────────────┐
                  │  transform_revenue  │  ← City-wise, product-wise aggregation
                  └──────────┬──────────┘
                             │
                             ▼
                  ┌─────────────────────┐
                  │  validate_output    │  ← Row count check, schema validation
                  └──────────┬──────────┘
                             │
                             ▼
                  ┌─────────────────────┐
                  │  load_to_warehouse  │  ← BigQuery / Redshift mein load
                  └──────────┬──────────┘
                             │
                             ▼
                  ┌─────────────────────┐
                  │  notify_team_slack  │  ← "Daily report ready" message
                  └─────────────────────┘
            

Is DAG mein cleaning steps parallel chalte hain — teeno simultaneously. Lekin transform_revenue tab tak nahi chalega jab tak teeno cleaning tasks complete nahi ho jaate. Yeh dependency tracking automatically handle hoti hai pipeline orchestrator ke through.

Agar validate_output fail ho jaaye — load_to_warehouse chalega hi nahi. Corrupt data warehouse mein pohunchega hi nahi. Aur team ko Slack pe alert milega ki validation fail hui.

Apache Airflow — Industry Standard Orchestrator

Apache Airflow Airbnb ne 2014 mein banaya tha apne internal data pipelines ke liye. 2016 mein open-source hua. Aaj yeh data engineering mein sabse widely used pipeline orchestration tool hai — Uber, Twitter, Lyft, Flipkart — sab use karte hain.

Airflow mein tum DAG ko Python code mein likhte ho. Yeh ek web UI deta hai jahan se tum apni pipelines monitor kar sakte ho, manually trigger kar sakte ho, ya failed tasks ko individually retry kar sakte ho.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

# Default settings — har task pe apply honge
default_args = {
    'owner': 'data-team',
    'retries': 3,                          # Failure pe 3 baar retry karega
    'retry_delay': timedelta(minutes=5),   # 5 minute wait karke retry
    'email_on_failure': True,              # Fail hone pe email
    'email': ['data-team@company.com']
}

# DAG define karo
with DAG(
    dag_id='daily_sales_pipeline',
    default_args=default_args,
    schedule_interval='0 2 * * *',         # Har raat 2 baje
    start_date=datetime(2025, 1, 1),
    catchup=False                          # Past missed runs ignore karo
) as dag:

    # Task 1 — Extract
    extract_task = PythonOperator(
        task_id='extract_raw_sales',
        python_callable=extract_from_s3    # Tumhara Python function
    )

    # Task 2 — Transform
    transform_task = PythonOperator(
        task_id='transform_revenue',
        python_callable=aggregate_by_city
    )

    # Task 3 — Load
    load_task = PythonOperator(
        task_id='load_to_warehouse',
        python_callable=load_to_bigquery
    )

    # Dependency define karo — yahi DAG banata hai
    extract_task >> transform_task >> load_task
    # Extract complete → Transform start → Load start
            

Airflow powerful hai lekin setup heavy hai. Tumhe ek dedicated server chahiye. Ek Postgres ya MySQL database chahiye. Aur ek scheduler process alag se chalti hai. Large teams ke liye yeh worthwhile hai — lekin small projects ke liye over-engineering ho sakta hai.

Prefect — Modern Python-Native Alternative

Prefect ek newer orchestration tool hai jo Airflow ki complexity ko address karne ke liye banaya gaya. Prefect ka main pitch yahi hai: "Yeh sirf Python hai." Alag DAG syntax nahi seekhna. Jo Python tumhe pata hai, wahi use karo — bas decorators lagao.

from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

# @task decorator — is function ko pipeline task banao
@task(retries=3, retry_delay_seconds=60, cache_key_fn=task_input_hash)
def extract_raw_sales():
    # S3 ya database se data pull karo
    return fetch_data_from_s3("s3://bucket/sales/")

@task
def transform_revenue(raw_data):
    # City-wise revenue aggregate karo
    return raw_data.groupby("city")["revenue"].sum()

@task
def load_to_warehouse(transformed_data):
    # BigQuery mein load karo
    write_to_bigquery(transformed_data, table="sales.daily_revenue")

# @flow decorator — yeh tera pipeline entry point hai
@flow(name="Daily Sales Pipeline")
def daily_sales_pipeline():
    raw = extract_raw_sales()
    aggregated = transform_revenue(raw)
    load_to_warehouse(aggregated)

if __name__ == "__main__":
    daily_sales_pipeline()        # Locally test kar sakte ho directly
            

Prefect mein tumhe koi alag configuration file nahi likhni. Python function likhte ho, decorator lagate ho — pipeline ready hai. Local pe bhi run kar sakte ho, production pe bhi. Testing bahut simpler hai kyunki yeh plain Python functions hain.

Manual Cron Scripts vs Data Pipeline — Production Mein Fark

Dono approaches ek simple scenario mein kaam karti hain. Fark tab samajh aata hai jab pipeline complex hoti hai, data volume badhta hai, ya kuch unexpected fail hota hai at 2 AM.

  CRON JOB APPROACH:
  ─────────────────────────────────────────────────────
  step_1.py  ✓   step_2.py  ✓   step_3.py  ✗   [SILENT]
                                     │
                                     └── Script exits with error code 1
                                         No one knows.
                                         step_4, step_5, step_6 — did they run?
                                         Was data partially loaded?
                                         Dashboard empty in the morning.
                                         Manual investigation starts at 9 AM.

  AIRFLOW / PREFECT PIPELINE:
  ─────────────────────────────────────────────────────
  Task 1  ✓   Task 2  ✓   Task 3  ✗
                               │
                               ├── Immediate Slack / Email alert: "Task 3 failed"
                               ├── Retry attempt 1 (after 5 min)  ✗
                               ├── Retry attempt 2 (after 5 min)  ✗
                               ├── Retry attempt 3 (after 5 min)  ✓  ← Fixed itself
                               │
                               ├── Task 4, 5, 6 resume automatically
                               └── Full logs available in UI — exact error line