Data Pipelines & ETL Basics for Data Analysts

import pandas as pd
import sqlalchemy
from google.cloud import bigquery
from datetime import datetime, timedelta
import logging

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s %(levelname)s %(message)s')
log = logging.getLogger(__name__)

# ── CONFIG ───────────────────────────────────────────────────
MYSQL_CONN   = 'mysql+pymysql://user:pass@db.company.in/orders_prod'
BQ_PROJECT   = 'evika-analytics'
BQ_DATASET   = 'ecommerce'
BQ_TABLE     = 'daily_orders'
LOOKBACK_HRS = 26  # pull last 26 hours — catches late-arriving data


# ── EXTRACT ──────────────────────────────────────────────────
def extract_orders(cutoff_utc: datetime) -> pd.DataFrame:
    """Pull orders created or updated since cutoff from MySQL."""
    engine = sqlalchemy.create_engine(MYSQL_CONN)
    query  = f"""
        SELECT
            o.order_id,
            o.customer_id,
            o.amount_inr,
            o.discount_inr,
            o.status,
            o.payment_method,
            o.city,
            o.state,
            o.created_at   AS order_created_at,
            o.updated_at   AS order_updated_at,
            c.tier         AS city_tier
        FROM orders o
        JOIN cities c ON o.city = c.city_name
        WHERE o.updated_at >= '{cutoff_utc.strftime('%Y-%m-%d %H:%M:%S')}'
        ORDER BY o.updated_at
    """
    df = pd.read_sql(query, engine)
    log.info(f"Extracted {len(df):,} rows from MySQL")
    return df


# ── VALIDATE ─────────────────────────────────────────────────
def validate(df: pd.DataFrame) -> pd.DataFrame:
    """Fail fast on data quality issues; log warnings for softer issues."""
    assert not df.empty, "No rows extracted — check source query or schedule"
    assert df['order_id'].is_unique, "Duplicate order_ids in extract"
    assert df['amount_inr'].ge(0).all(), "Negative order amounts found"

    null_pct = df.isnull().mean()
    for col, pct in null_pct.items():
        if pct > 0.05:
            log.warning(f"Column '{col}' has {pct:.1%} nulls")

    log.info(f"Validation passed: {len(df):,} rows, {df['order_id'].nunique():,} unique orders")
    return df


# ── TRANSFORM ────────────────────────────────────────────────
def transform(df: pd.DataFrame) -> pd.DataFrame:
    """Clean and enrich the extracted data."""
    # Timestamps: source is IST, store as UTC in warehouse
    for col in ['order_created_at', 'order_updated_at']:
        df[col] = (pd.to_datetime(df[col])
                     .dt.tz_localize('Asia/Kolkata')
                     .dt.tz_convert('UTC'))

    # Derived fields
    df['net_amount_inr'] = df['amount_inr'] - df['discount_inr'].fillna(0)
    df['order_date']     = df['order_created_at'].dt.date          # IST date
    df['order_hour_ist'] = (df['order_created_at']
                              .dt.tz_convert('Asia/Kolkata').dt.hour)

    # Standardise categoricals
    df['payment_method'] = df['payment_method'].str.upper().str.strip()
    df['state']          = df['state'].str.title().str.strip()

    # Financial year (April–March, Indian FY)
    def fy(d):
        return d.year if d.month >= 4 else d.year - 1
    df['financial_year'] = df['order_date'].apply(
        lambda d: f"FY{fy(d)}-{str(fy(d)+1)[-2:]}"
    )

    # Flag Diwali orders (hardcoded for 2026; automate with holiday library)
    diwali_2026 = pd.to_datetime('2026-11-08').date()
    df['is_diwali_window'] = df['order_date'].apply(
        lambda d: abs((d - diwali_2026).days) <= 7
    )

    log.info(f"Transform complete: {len(df):,} rows, ₹{df['net_amount_inr'].sum()/1e7:.2f}Cr net GMV")
    return df


# ── LOAD ─────────────────────────────────────────────────────
def load_to_bigquery(df: pd.DataFrame, date_partition: str):
    """UPSERT: delete the partition date first, then insert. Idempotent."""
    client = bigquery.Client(project=BQ_PROJECT)
    table_ref = f"{BQ_PROJECT}.{BQ_DATASET}.{BQ_TABLE}"

    # Delete today's partition so reruns do not duplicate
    delete_sql = f"""
        DELETE FROM `{table_ref}`
        WHERE order_date = '{date_partition}'
    """
    client.query(delete_sql).result()
    log.info(f"Deleted existing rows for {date_partition}")

    # Load
    job_config = bigquery.LoadJobConfig(
        write_disposition = bigquery.WriteDisposition.WRITE_APPEND,
        time_partitioning = bigquery.TimePartitioning(field='order_date'),
        clustering_fields = ['city', 'payment_method', 'status'],
        schema = [
            bigquery.SchemaField('order_id',         'STRING'),
            bigquery.SchemaField('customer_id',      'STRING'),
            bigquery.SchemaField('amount_inr',       'FLOAT64'),
            bigquery.SchemaField('net_amount_inr',   'FLOAT64'),
            bigquery.SchemaField('order_date',       'DATE'),
            bigquery.SchemaField('financial_year',   'STRING'),
            bigquery.SchemaField('is_diwali_window', 'BOOL'),
        ],
    )
    job = client.load_table_from_dataframe(df, table_ref, job_config=job_config)
    job.result()
    log.info(f"Loaded {len(df):,} rows to {table_ref}")


# ── ORCHESTRATE ──────────────────────────────────────────────
def run_pipeline():
    cutoff   = datetime.utcnow() - timedelta(hours=LOOKBACK_HRS)
    date_str = (datetime.utcnow() - timedelta(hours=5, minutes=30)).strftime('%Y-%m-%d')

    raw       = extract_orders(cutoff)
    validated = validate(raw)
    clean     = transform(validated)
    load_to_bigquery(clean, date_str)
    log.info("Pipeline complete ✅")

if __name__ == '__main__':
    run_pipeline()

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.email  import EmailOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime, timedelta

# ── Default arguments applied to every task ──────────────────
default_args = {
    'owner':            'data-team',
    'retries':          2,                        # retry failed tasks up to 2 times
    'retry_delay':      timedelta(minutes=10),    # wait 10 min between retries
    'email_on_failure': True,
    'email':            ['data-alerts@company.in'],
}

# ── DAG Definition ───────────────────────────────────────────
with DAG(
    dag_id          = 'daily_orders_pipeline',
    description     = 'Extract orders from MySQL → BigQuery, daily at 2:30 AM IST',
    schedule_interval = '0 21 * * *',   # 21:00 UTC = 02:30 IST
    start_date      = datetime(2026, 9, 1),
    catchup         = False,            # do not backfill missed runs
    default_args    = default_args,
    tags            = ['orders', 'production', 'bigquery'],
) as dag:

    # Task 1: Extract from MySQL
    extract = PythonOperator(
        task_id         = 'extract_orders_mysql',
        python_callable = extract_orders,
        op_kwargs       = {'cutoff_utc': '{{ data_interval_start }}'},
    )

    # Task 2: Validate
    validate_task = PythonOperator(
        task_id         = 'validate_data',
        python_callable = validate,
    )

    # Task 3: Transform
    transform_task = PythonOperator(
        task_id         = 'transform_orders',
        python_callable = transform,
    )

    # Task 4: Load to BigQuery
    load = PythonOperator(
        task_id         = 'load_bigquery',
        python_callable = load_to_bigquery,
        op_kwargs       = {'date_partition': '{{ ds }}'},
    )

    # Task 5: Refresh Power BI / dbt model
    refresh_dbt = BashOperator(
        task_id  = 'run_dbt_models',
        bash_command = 'dbt run --select orders+ --profiles-dir /opt/dbt',
    )

    # Task 6: Alert on completion
    notify = EmailOperator(
        task_id   = 'notify_success',
        to        = ['analytics@company.in'],
        subject   = 'Daily orders pipeline complete — {{ ds }}',
        html_content = '<p>Orders for {{ ds }} loaded to BigQuery. ✅</p>',
    )

    # ── Task dependencies (left → right = must run before) ───
    extract >> validate_task >> transform_task >> load >> refresh_dbt >> notify
    #         ↑ validate runs only after extract succeeds
    #                          ↑ load runs only after transform succeeds
    #                                          ↑ dbt runs only after load succeeds

What is the difference between ETL and ELT?

ETL (Extract, Transform, Load) transforms data before loading it into the destination. Raw data comes from source systems, is cleaned and restructured in a staging area or dedicated transformation layer, and only the processed data lands in the data warehouse. ELT (Extract, Load, Transform) loads raw data first into the destination, then transforms it there using SQL or dbt. ETL was the dominant approach when storage was expensive and warehouse compute was limited — you did not want to store messy raw data. ELT has become the modern standard because cloud data warehouses (BigQuery, Snowflake, Redshift) offer cheap storage and powerful parallel SQL compute, making in-warehouse transformation faster and more flexible. In Indian company tech stacks: legacy on-premise systems (banks, manufacturing, government) typically use ETL with tools like Informatica or SSIS. Modern startups and e-commerce companies use ELT with tools like dbt + BigQuery or Snowflake.

What is a data warehouse and how is it different from a data lake?

A data warehouse stores structured, processed, analysis-ready data in a predefined schema. It is optimised for fast SQL queries by analysts and BI tools. Examples: Google BigQuery, Snowflake, Amazon Redshift, AWS Redshift. Strengths: fast query performance, enforced data quality, governed and trusted. Weakness: less flexible — schema must be defined upfront; storing unstructured data (images, logs, free text) is difficult or expensive. A data lake stores raw data of any format — structured tables, JSON logs, CSV exports, images, audio, documents — at very low cost, typically on object storage (AWS S3, Google Cloud Storage, Azure Blob). Strengths: stores everything; handles unstructured data; retains raw history. Weakness: without governance, becomes a "data swamp" — everything is in there but nothing is easy to find or trust. A data lakehouse (Delta Lake, Apache Iceberg) combines both: lake-style storage with warehouse-style governance and query performance. Most Indian companies at scale use a combination: raw data in S3 / GCS, processed data in BigQuery or Redshift, with a dbt transformation layer in between.

Do data analysts in India need to know Apache Airflow?

For data analyst roles in India, a conceptual understanding of Airflow is increasingly expected — you should be able to read a DAG, understand what triggers a pipeline, and diagnose a failed task. Building Airflow DAGs from scratch is more of a data engineer responsibility. However, at many Indian startups and growth-stage companies, the data analyst and data engineer roles overlap, and analysts are expected to write and maintain simple Airflow DAGs (daily report pipelines, data refresh jobs). Being able to write a basic DAG sets you apart in interviews and is achievable with a week of practice. Other orchestration tools common in Indian companies: Apache Airflow (most common open-source), Prefect and Dagster (newer, more Python-native), dbt Cloud (for transformation scheduling), and cloud-native options (AWS Glue, Google Cloud Composer which is managed Airflow, Azure Data Factory).

What are the most common data pipeline failures and how do you fix them?

The five most common pipeline failures in Indian data stacks: (1) Source schema changes — an upstream team adds or renames a column and the pipeline breaks because the transformation code references the old name. Fix: schema validation at extraction, alerting on schema drift, flexible extraction that captures all columns. (2) Volume anomalies — daily orders spike on Diwali to 10x normal; the pipeline times out or runs out of memory. Fix: incremental loads instead of full refreshes; adaptive batch sizes; partitioned processing. (3) Duplicate records — pipeline reruns after a failure and inserts data twice. Fix: idempotent writes (INSERT IGNORE, MERGE/UPSERT, or delete-then-insert for the affected partition). (4) Late-arriving data — vendor sends previous day's transactions 2 hours after midnight; the 1 AM pipeline misses them. Fix: watermark-based loading with a 24-hour lookback window for recent dates. (5) Timezone issues — API timestamps in UTC, database in IST (UTC+5:30), dashboard shows wrong day boundaries. Fix: always store timestamps in UTC, convert to IST only at display layer.