Data Pipeline Architecture – ETL vs ELT, Orchestration & Best Practices 2026

Every data science project depends on reliable data pipelines. A pipeline that breaks silently — delivering stale or incorrect data — is worse than no pipeline at all. This guide covers modern data pipeline architecture: when to use ETL vs ELT, how to orchestrate with Airflow, and how to build pipelines that are reliable, testable, and maintainable in production.

ETL vs ELT

ETL (Extract, Transform, Load) transforms data before loading it into the warehouse. ELT (Extract, Load, Transform) loads raw data first, then transforms it inside the warehouse using SQL. ELT is the modern approach for cloud data warehouses (BigQuery, Snowflake, Redshift) because compute inside the warehouse is cheap and scalable, and keeping raw data enables reprocessing when transformation logic changes. Use ETL when you have strict data privacy requirements or when transformation is too complex for SQL.

Building an ETL Pipeline in Python

import pandas as pd
import sqlalchemy as sa
import logging
from datetime import datetime, timedelta

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

class ETLPipeline:
    def __init__(self, source_conn: str, target_conn: str):
        self.source = sa.create_engine(source_conn)
        self.target = sa.create_engine(target_conn)

    def extract(self, table: str, since: datetime) -> pd.DataFrame:
        logger.info(f'Extracting {table} since {since}')
        query = f'''
            SELECT * FROM {table}
            WHERE updated_at >= :since
        '''
        df = pd.read_sql(query, self.source, params={'since': since})
        logger.info(f'Extracted {len(df)} rows')
        return df

    def transform(self, df: pd.DataFrame) -> pd.DataFrame:
        logger.info('Transforming data')
        df = df.drop_duplicates(subset=['id'])
        df['amount'] = pd.to_numeric(df['amount'], errors='coerce')
        df = df.dropna(subset=['id', 'amount'])
        df['amount_usd']   = df['amount'] * 0.012
        df['processed_at'] = datetime.utcnow()
        df['date']         = pd.to_datetime(df['created_at']).dt.date
        return df

    def validate(self, df: pd.DataFrame) -> bool:
        assert df['id'].is_unique,    'Duplicate IDs found'
        assert df['amount'].ge(0).all(), 'Negative amounts found'
        assert df['amount_usd'].notna().all(), 'Null amounts found'
        logger.info(f'Validation passed — {len(df)} rows')
        return True

    def load(self, df: pd.DataFrame, table: str) -> None:
        logger.info(f'Loading {len(df)} rows into {table}')
        df.to_sql(table, self.target, if_exists='append',
                  index=False, chunksize=1000,
                  method='multi')
        logger.info('Load complete')

    def run(self, table: str, days_back: int = 1) -> None:
        since = datetime.utcnow() - timedelta(days=days_back)
        df    = self.extract(table, since)
        df    = self.transform(df)
        if self.validate(df):
            self.load(df, f'processed_{table}')

pipeline = ETLPipeline(
    source_conn='postgresql://user:pass@source/db',
    target_conn='postgresql://user:pass@warehouse/db'
)
pipeline.run('transactions', days_back=1)

Orchestrating with Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
from datetime import timedelta

default_args = {
    'owner':            'data-team',
    'retries':          3,
    'retry_delay':      timedelta(minutes=5),
    'email_on_failure': True,
    'email':            ['alerts@dataexpertise.in'],
}

with DAG(
    dag_id='daily_etl_pipeline',
    default_args=default_args,
    schedule_interval='0 6 * * *',   # 6am daily
    start_date=days_ago(1),
    catchup=False,
    tags=['etl', 'production'],
) as dag:

    def extract_task(**context):
        # Pull execution date from context for idempotent extraction
        exec_date = context['execution_date']
        logger.info(f'Extracting data for {exec_date}')
        # ... extraction logic ...
        return len(df)   # return value stored in XCom

    def transform_task(**context):
        row_count = context['ti'].xcom_pull(task_ids='extract')
        logger.info(f'Transforming {row_count} rows')
        # ... transformation logic ...

    def validate_and_load(**context):
        # ... validation and load logic ...
        pass

    extract   = PythonOperator(task_id='extract',   python_callable=extract_task)
    transform = PythonOperator(task_id='transform', python_callable=transform_task)
    load      = PythonOperator(task_id='load',      python_callable=validate_and_load)
    notify    = BashOperator(task_id='notify',
                             bash_command='echo "Pipeline complete"')

    extract >> transform >> load >> notify

dbt for ELT Transformations

-- models/staging/stg_transactions.sql
WITH source AS (
    SELECT * FROM {{ source('raw', 'transactions') }}
),
cleaned AS (
    SELECT
        id,
        user_id,
        CAST(amount AS DECIMAL(18,2))      AS amount,
        CAST(amount * 0.012 AS DECIMAL(18,2)) AS amount_usd,
        CAST(created_at AS TIMESTAMP)      AS created_at,
        DATE(created_at)                   AS transaction_date,
        LOWER(TRIM(status))               AS status,
        CURRENT_TIMESTAMP                  AS processed_at
    FROM source
    WHERE id IS NOT NULL
      AND amount > 0
      AND status NOT IN ('cancelled', 'test')
)
SELECT * FROM cleaned

-- models/marts/revenue_daily.sql
{{ config(materialized='table', partition_by={'field':'transaction_date'}) }}

SELECT
    transaction_date,
    COUNT(*)          AS num_transactions,
    SUM(amount_usd)   AS total_revenue_usd,
    AVG(amount_usd)   AS avg_transaction_usd,
    COUNT(DISTINCT user_id) AS unique_users
FROM {{ ref('stg_transactions') }}
WHERE status = 'completed'
GROUP BY transaction_date

# Run dbt
dbt run --models staging.stg_transactions marts.revenue_daily
dbt test    # run data quality tests
dbt docs generate && dbt docs serve   # generate documentation

Data Quality with Great Expectations

import great_expectations as ge

df_ge = ge.from_pandas(df)

# Define expectations
df_ge.expect_column_to_exist('id')
df_ge.expect_column_values_to_be_unique('id')
df_ge.expect_column_values_to_not_be_null('amount')
df_ge.expect_column_values_to_be_between('amount', 0, 1_000_000)
df_ge.expect_column_values_to_be_in_set('status',
    ['completed', 'pending', 'failed'])

# Validate
results = df_ge.validate()
if not results['success']:
    failed = [r for r in results['results'] if not r['success']]
    raise ValueError(f'Data quality failed: {len(failed)} checks')

Pipeline Best Practices

Make every pipeline idempotent — running it twice should produce the same result. Use incremental loads rather than full refreshes wherever possible. Log row counts at each stage and alert when counts deviate more than 20% from the previous run. Store raw data permanently before transforming — you will need to reprocess. Version your transformation logic with git. Test pipelines with real production data (anonymised) before deploying, not synthetic data that may not reflect edge cases.

Conclusion

Reliable data pipelines are the infrastructure on which all data science value is built. Invest in proper orchestration with Airflow, declarative transformations with dbt, and automated data quality checks with Great Expectations. A pipeline that runs reliably at 6am every day without manual intervention is worth more than any individual analysis built on ad-hoc data pulls.