Why Every Data Engineer Should Learn Python in 2026

from pathlib import Path
import pandas as pd

# Combine every daily CSV in a folder into one DataFrame
raw_dir = Path("raw/orders")
frames = []
for f in sorted(raw_dir.glob("orders_*.csv")):
    df = pd.read_csv(f, dtype={"order_id": "string"}, parse_dates=["created_at"])
    df["source_file"] = f.name          # keep lineage
    frames.append(df)

orders = pd.concat(frames, ignore_index=True)
print(orders.shape, orders["source_file"].nunique(), "files")

# Save as Parquet: typed, compressed, fast to read
orders.to_parquet("staging/orders.parquet", index=False)

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()
retry = Retry(total=5, backoff_factor=1,
              status_forcelist=[429, 500, 502, 503, 504])
session.mount("https://", HTTPAdapter(max_retries=retry))

def fetch_all(url, token):
    rows, page = [], 1
    while True:
        r = session.get(url, params={"page": page, "per_page": 100},
                        headers={"Authorization": f"Bearer {token}"},
                        timeout=30)
        r.raise_for_status()
        batch = r.json()
        if not batch:
            break
        rows.extend(batch)
        page += 1
    return rows

import os
import pandas as pd
from sqlalchemy import create_engine, text

engine = create_engine(os.environ["WAREHOUSE_URL"])  # never hard-code passwords

def transform(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df["email"] = df["email"].str.strip().str.lower()
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
    df["order_date"] = pd.to_datetime(df["created_at"]).dt.date
    return df.drop_duplicates(subset="order_id")

def validate(df: pd.DataFrame) -> None:
    problems = []
    if df["order_id"].isna().any():
        problems.append("null order_id")
    if df["amount"].isna().any() or (df["amount"] < 0).any():
        problems.append("missing or negative amount")
    if problems:
        raise ValueError("Validation failed: " + ", ".join(problems))

def load(df: pd.DataFrame, day: str) -> None:
    cols = ["order_id", "email", "amount", "order_date"]
    with engine.begin() as conn:  # one transaction
        conn.execute(text("DELETE FROM orders WHERE order_date = :d"), {"d": day})
        df[cols].to_sql("orders", conn, if_exists="append", index=False)

def run(day: str) -> None:
    raw = pd.read_parquet(f"staging/orders_{day}.parquet")
    clean = transform(raw)
    validate(clean)
    load(clean, day)
    print(f"{day}: loaded {len(clean)} rows")

import pandas as pd
import pytest
from etl_orders import transform, validate

def sample():
    return pd.DataFrame({
        "order_id": ["A1", "A1", "A2"],
        "email": [" User@Mail.com ", " User@Mail.com ", "x@y.in"],
        "amount": ["250", "250", "99.5"],
        "created_at": ["2026-03-01 10:00"] * 3,
    })

def test_transform_dedupes_and_cleans():
    out = transform(sample())
    assert len(out) == 2
    assert out["email"].iloc[0] == "user@mail.com"

def test_validate_rejects_negative_amount():
    df = transform(sample())
    df.loc[0, "amount"] = -5
    with pytest.raises(ValueError):
        validate(df)

# Run with:  pytest -q

from datetime import datetime, timedelta
from airflow.decorators import dag, task

@dag(schedule="@daily", start_date=datetime(2026, 1, 1),
     catchup=False, default_args={"retries": 2,
     "retry_delay": timedelta(minutes=10)}, tags=["orders"])
def orders_pipeline():

    @task
    def extract(ds=None):
        # ds is the run date, e.g. "2026-03-01"
        from extract import save_orders_for_day  # src/extract.py
        return save_orders_for_day(ds)  # returns staging file path

    @task
    def transform_and_load(path: str, ds=None):
        from etl_orders import run  # the job from section 02
        run(ds)

    transform_and_load(extract())

orders_pipeline()

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("daily_revenue").getOrCreate()

orders = spark.read.parquet("s3a://lake/orders/")

daily = (orders
    .filter(F.col("amount") > 0)
    .withColumn("order_date", F.to_date("created_at"))
    .groupBy("order_date", "city")
    .agg(F.sum("amount").alias("revenue"),
         F.countDistinct("customer_id").alias("customers")))

(daily.write
    .mode("overwrite")
    .partitionBy("order_date")
    .parquet("s3a://lake/marts/daily_revenue/"))

spark.stop()

orders-pipeline/
├── README.md            # what it does, how to run, sample output
├── pyproject.toml       # dependencies pinned
├── .env.example         # variable names only, no secrets
├── src/
│   ├── extract.py       # API and file readers
│   ├── transform.py     # pure functions, easy to test
│   ├── validate.py      # quality rules
│   └── load.py          # idempotent writes
├── dags/
│   └── orders_dag.py    # Airflow wiring only
├── tests/
│   └── test_transform.py
└── sql/
    └── create_tables.sql

PIPELINE QUALITY CHECKLIST

[ ] Every network call has a timeout and retries
[ ] Secrets come from env vars or a secrets manager
[ ] Explicit dtypes when reading files
[ ] Row counts logged after extract and after load
[ ] Primary key checked for nulls and duplicates
[ ] Business rules checked (amount >= 0, valid dates)
[ ] Load is idempotent (rerun gives same result)
[ ] Failures raise errors and trigger an alert
[ ] Unit tests for transform and validate
[ ] README explains schedule, inputs and outputs

PYTHON FOR DATA ENGINEERING — 8 WEEK PLAN

Weeks 1-2 : Core Python — functions, dicts, files,
            exceptions, virtual environments, Git
Weeks 3-4 : pandas — read_csv/parquet, merge, groupby,
            dtypes, dates; plus SQL joins and windows
Week 5    : APIs with requests, JSON, pagination,
            retries; load into PostgreSQL via SQLAlchemy
Week 6    : pytest for transform and validate functions
Week 7    : Airflow locally — one DAG, retries, backfill
Week 8    : PySpark basics on a sample dataset;
            publish the full project on GitHub