SQL is the language of the warehouse; Python is the language of everything around it: pulling from APIs, reading awkward files, calling the warehouse, transforming what SQL cannot express, and gluing steps into a pipeline. The data engineer's Python is not the data scientist's notebook. It runs unattended, on a schedule, against data that changes shape, and it must be idempotent, typed enough to fail early, and honest about memory. This module covers the DataFrame libraries you will actually use (pandas and its faster successor polars), typed records, files and formats, calling APIs and databases robustly, and the structure of a script that can be re-run without fear.
- Load, clean and reshape tabular data with pandas and polars, and know when each is the right tool
- Use typed records (dataclasses, pydantic) to validate data at the boundary
- Read and write CSV, JSON and Parquet correctly, including types, encodings and compression
- Call APIs and databases with retries, pagination and parameterised queries
- Structure a pipeline script so it is idempotent, testable and observable
DataFrames: pandas and polars
A DataFrame is an in-memory table with typed columns. pandas is the long-standing standard: enormous ecosystem, every tutorial, single-threaded and memory-hungry (a rule of thumb is five to ten times the file size in RAM). polars is the modern alternative: multi-threaded, a lazy query engine that optimises whole pipelines, a stricter and more consistent API, and typically several times faster with far less memory. For new pipeline code polars is the better default; pandas remains necessary where libraries expect it and is fine for small data.
import pandas as pd
import polars as pl
# pandas: eager, one step at a time
orders = pd.read_csv("orders.csv", dtype={"order_id": "int64", "customer_id": "int64"},
parse_dates=["ordered_at"])
revenue_pd = (
orders[orders["status"] == "paid"]
.assign(month=lambda d: d["ordered_at"].dt.to_period("M").dt.to_timestamp())
.groupby("month", as_index=False)
.agg(revenue=("amount", "sum"), orders=("order_id", "nunique"))
)
# polars: lazy plan, executed once, in parallel
revenue_pl = (
pl.scan_csv("orders.csv", schema_overrides={"order_id": pl.Int64, "customer_id": pl.Int64},
try_parse_dates=True)
.filter(pl.col("status") == "paid")
.with_columns(pl.col("ordered_at").dt.truncate("1mo").alias("month"))
.group_by("month")
.agg(pl.col("amount").sum().alias("revenue"), pl.col("order_id").n_unique().alias("orders"))
.sort("month")
.collect()
)Three habits apply to both. Set dtypes explicitly at load time; inferred types drift between files (an id column that becomes float because one row is empty is a classic). Chain transformations as expressions rather than mutating in place, so each step is inspectable and the code reads as a pipeline. And profile memory before scaling up: df.memory_usage(deep=True) in pandas, df.estimated_size() in polars. When data does not fit, the answer is usually not a bigger machine but pushing the work into the warehouse or a distributed engine, which later modules cover.
For local analytics on files, DuckDB reads CSV and Parquet directly with SQL, joins them, and returns pandas or polars frames. Many "which library" questions dissolve into "write the SQL in DuckDB and keep Python for the glue".
Types at the boundary
Data enters your code from files and APIs whose shape you do not control. Validate it at the boundary, once, into typed records, so that everything downstream can assume clean data. Python's dataclasses give structure; pydantic adds validation and coercion ("the string "12.50" becomes a Decimal, an invalid email fails with a clear message"). Rejecting or quarantining a bad record at ingestion, with a reason, is far cheaper than discovering it as a NULL in a report.
from datetime import datetime
from decimal import Decimal
from typing import Literal
from pydantic import BaseModel, EmailStr, Field, ValidationError
class OrderRecord(BaseModel):
order_id: int
customer_email: EmailStr
ordered_at: datetime
status: Literal["placed", "paid", "shipped", "refunded"]
amount: Decimal = Field(ge=0, max_digits=12, decimal_places=2)
model_config = {"extra": "forbid"} # unexpected columns are a schema change, not noise
def parse_rows(rows: list[dict]) -> tuple[list[OrderRecord], list[dict]]:
good, bad = [], []
for row in rows:
try:
good.append(OrderRecord.model_validate(row))
except ValidationError as e:
bad.append({"row": row, "errors": e.errors()})
return good, badUse Decimal for money and datetime with an explicit timezone for timestamps; floats and naive datetimes are the two most common sources of silently wrong numbers. When a schema changes upstream (a new column, a renamed field), extra: forbid turns it into an immediate, attributable failure instead of a quiet drift.
Files and formats
CSV is universal and terrible: no types, no schema, ambiguous quoting, encodings and line endings that vary by source. Always specify the encoding, the delimiter, the dtypes and how NULLs are spelled; treat every CSV as untrusted input. JSON carries structure and types but not tabular shape; newline-delimited JSON (one object per line) streams and splits well, while a single giant array does not. Parquet is the format to write: columnar, typed, compressed, with embedded schema and statistics that let readers skip data. Convert to Parquet as early as possible and keep the raw file for provenance.
import polars as pl
raw = pl.read_csv(
"export.csv",
encoding="utf8-lossy", # replace undecodable bytes instead of crashing
separator=";",
null_values=["", "NULL", "N/A", "-"],
schema_overrides={"customer_id": pl.Int64, "amount": pl.Decimal(12, 2)},
try_parse_dates=True,
infer_schema_length=10_000, # look further before guessing types
)
(raw.with_columns(pl.col("ordered_at").dt.strftime("%Y-%m").alias("month"))
.write_parquet("orders/", partition_by="month", compression="zstd"))
# downstream reads only the months it needs; the filter is pushed into the file scan
q = pl.scan_parquet("orders/**/*.parquet").filter(pl.col("month") == "2026-03")Large files should be processed in chunks or lazily rather than loaded whole: pl.scan_* builds a plan and streams; pandas read_csv(chunksize=...) iterates. Compression choice matters less than format choice; zstd or snappy for Parquet is a good default. The storage module goes deeper into layouts, partitioning and table formats.
APIs and databases, robustly
Unattended code meets unreliable networks. Every HTTP call gets a timeout, retries with exponential backoff on transient failures (429, 5xx, connection errors) and no retry on client errors (4xx other than 429). Paginate with the API's cursor, not by guessing page numbers, and persist the cursor so a restart resumes rather than repeats. Respect rate limits from headers. For databases, use parameterised queries always, stream large results with server-side cursors or batches, and keep transactions short.
import time
from collections.abc import Iterator
import httpx
RETRYABLE = {429, 500, 502, 503, 504}
def fetch_all(client: httpx.Client, url: str, params: dict) -> Iterator[dict]:
cursor = None
while True:
attempt = 0
while True:
try:
r = client.get(url, params={**params, "cursor": cursor} if cursor else params, timeout=30.0)
if r.status_code in RETRYABLE:
raise httpx.HTTPStatusError("retryable", request=r.request, response=r)
r.raise_for_status()
break
except (httpx.TransportError, httpx.HTTPStatusError) as e:
attempt += 1
if attempt > 5 or (isinstance(e, httpx.HTTPStatusError) and e.response.status_code not in RETRYABLE):
raise
time.sleep(min(2 ** attempt, 60))
body = r.json()
yield from body["items"]
cursor = body.get("next_cursor")
if not cursor:
returnimport psycopg
UPSERT = """
INSERT INTO orders (order_id, customer_email, ordered_at, status, amount)
VALUES (%(order_id)s, %(customer_email)s, %(ordered_at)s, %(status)s, %(amount)s)
ON CONFLICT (order_id) DO UPDATE
SET status = EXCLUDED.status, amount = EXCLUDED.amount
"""
def load(records: list[dict], dsn: str, batch: int = 1000) -> int:
n = 0
with psycopg.connect(dsn) as conn:
for i in range(0, len(records), batch):
with conn.transaction():
with conn.cursor() as cur:
cur.executemany(UPSERT, records[i:i + batch])
n += len(records[i:i + batch])
return nThe shape of a script you can re-run
A pipeline step will be re-run: after a failure, for a backfill, by someone testing. Idempotency means running it twice produces the same result as once: upserts instead of inserts, overwrite-the-partition instead of append, deterministic output paths derived from the run's logical date rather than from now(). Take the logical date as a parameter; never compute "yesterday" inside the script, because a re-run on Thursday for Monday's data must process Monday.
import argparse
import logging
import sys
from datetime import date
log = logging.getLogger("orders_load")
def extract(day: date) -> list[dict]: ...
def transform(rows: list[dict]) -> tuple[list[dict], list[dict]]: ...
def load_partition(day: date, rows: list[dict]) -> int: ...
def run(day: date) -> int:
rows = extract(day)
good, bad = transform(rows)
if bad:
log.warning("quarantined %d bad rows for %s", len(bad), day)
n = load_partition(day, good) # overwrites the day's partition: idempotent
log.info("loaded %d rows for %s", n, day)
return n
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")
p = argparse.ArgumentParser()
p.add_argument("--day", type=date.fromisoformat, required=True, help="logical date to process")
args = p.parse_args()
try:
run(args.day)
except Exception:
log.exception("failed for %s", args.day)
sys.exit(1)Log counts at every stage (rows in, rows rejected, rows out) so a silent drop is visible; emit them as metrics if you have a metrics system. Configuration and secrets come from the environment or a secrets manager, never from the code. Package the pipeline as a proper project with pinned dependencies and tests for transform using small fixture files. The orchestration module takes this script and schedules it; everything it needs — parameters, idempotency, exit codes, logs — is already here.