SQL vs Python for Data Engineers A Practical Comparison

-- models/stg_orders_deduped.sql
WITH ranked AS (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY order_id
                              ORDER BY _loaded_at DESC) AS rn
    FROM {{ source('shop', 'raw_orders') }}
)
SELECT order_id, customer_id, order_date, amount, status
FROM ranked
WHERE rn = 1

import pandas as pd

raw = pd.read_parquet("raw_orders.parquet")
deduped = (raw.sort_values("_loaded_at", ascending=False)
              .drop_duplicates(subset="order_id", keep="first")
              [["order_id", "customer_id", "order_date", "amount", "status"]])

# Works for small data, but requires pulling the
# whole table out of the warehouse first.

import time
import requests

def fetch_all(url, token, page_size=100, max_retries=3):
    page, rows = 1, []
    while True:
        for attempt in range(max_retries):
            r = requests.get(url, params={"page": page, "per_page": page_size},
                             headers={"Authorization": f"Bearer {token}"}, timeout=30)
            if r.status_code == 429:            # rate limited
                time.sleep(2 ** attempt)
                continue
            r.raise_for_status()
            break
        else:
            raise RuntimeError(f"page {page} failed after {max_retries} retries")
        batch = r.json()
        if not batch:
            return rows
        rows.extend(batch)
        page += 1

from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator

with DAG("orders_elt", start_date=datetime(2026, 1, 1),
         schedule="@daily", catchup=False,
         default_args={"retries": 2}) as dag:

    ingest = PythonOperator(task_id="ingest_orders",
                            python_callable=lambda: print("load raw_orders"))

    transform = BashOperator(task_id="dbt_run",
                             bash_command="cd /opt/dbt && dbt run --select stg_orders_deduped+")

    test = BashOperator(task_id="dbt_test",
                        bash_command="cd /opt/dbt && dbt test --select stg_orders_deduped+")

    ingest >> transform >> test

# models/schema.yml
version: 2
models:
  - name: stg_orders_deduped
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: status
        tests:
          - accepted_values:
              values: ["PLACED", "SHIPPED", "DELIVERED", "CANCELLED"]

# test_transform.py
from transform import normalise_status

def test_known_status_is_uppercased():
    assert normalise_status(" shipped ") == "SHIPPED"

def test_unknown_status_maps_to_other():
    assert normalise_status("lost-in-transit") == "OTHER"

# transform.py
VALID = {"PLACED", "SHIPPED", "DELIVERED", "CANCELLED"}
def normalise_status(s: str) -> str:
    s = s.strip().upper()
    return s if s in VALID else "OTHER"

Pipeline step                         | Language
--------------------------------------|----------
Call an API / read odd files          | Python
Load raw data into warehouse          | Python (or a connector)
Deduplicate / join / aggregate        | SQL (dbt models)
Complex business rule with branches   | Python (tested), or SQL if simple
Data quality checks on tables         | SQL tests (dbt)
Scheduling, retries, alerts           | Python orchestrator
ML feature engineering at scale       | SQL or Spark, depending on platform

[ ] No warehouse-to-pandas round trips for simple joins
[ ] API code handles pagination, retries and timeouts
[ ] Every model has unique/not_null tests on its key
[ ] Python functions with logic have unit tests
[ ] Secrets come from environment or a secret store
[ ] DAG retries and alerts are configured

Mini ELT project
1. Python: ingest a paginated public API into raw table
2. SQL (dbt): staging model with dedup + typed columns
3. SQL (dbt): mart model with daily aggregates
4. Tests: dbt unique/not_null + pytest for helpers
5. Airflow: DAG ingest >> dbt run >> dbt test