Data Engineering for AI: Pipelines, Quality, Labeling, Features and Privacy

Models are only as good as the data that reaches them, and in production most ML incidents trace back to data: a renamed column, a silently changed unit, a label that leaked from the future, a batch that arrived twice. Data engineering for AI is the work of moving data reliably from where it is produced to where models train and predict, while keeping it correct, traceable, lawful and documented. It is often the largest part of an ML team's effort.

Intuition: a water utility

Think of a city water system. Water is collected from many sources (applications, sensors, logs, vendors), carried through pipes (pipelines), treated (cleaned, validated, de-identified), stored in reservoirs (warehouses and lakes), and delivered to taps (training jobs and online models). Treatment plants test samples constantly, because by the time contamination reaches a tap it is too late. Every pipe has a map so engineers can trace a problem upstream. And there is a published water-quality report so consumers know what they are drinking.

Each piece has a counterpart in this lesson: pipelines, quality checks, storage, serving, lineage, and dataset documentation.

Pipelines: batch and streaming

Batch pipelines process bounded chunks of data on a schedule: "every night, compute yesterday's features". They are simpler, cheaper and easier to backfill. Streaming pipelines process unbounded event streams continuously with low latency: "update a user's fraud features within seconds of each transaction". Typical building blocks: event logs such as Apache Kafka; processing engines such as Apache Spark (batch and micro-batch) and Apache Flink (stream-first); orchestrators such as Apache Airflow or Dagster that schedule and monitor pipeline steps.

Many systems combine both: streaming for fresh online features and batch for training and reconciliation. The danger is two separate code paths computing "the same" feature differently, which causes training-serving skew.

ETL vs ELT

• ETL (extract, transform, load): transform before loading into the warehouse. Used when storage was expensive or data must be cleaned or de-identified before landing.
• ELT (extract, load, transform): load raw data into a scalable warehouse or lakehouse first, then transform with SQL (tools like dbt). Keeps raw history for reprocessing; now the common default.

For AI, a layered layout is common: raw (immutable, as received), cleaned/conformed (typed, deduplicated, validated), and curated (features, training sets). Sensitive data may need to be redacted before the raw layer if policy forbids storing it.

Pipeline engineering principles

• Idempotency: re-running a job for the same partition produces the same result and no duplicates (overwrite the partition, or upsert on a key).
• Partitioning by time so that backfills and reprocessing are cheap and targeted.
• Event time vs processing time: compute features by when events happened, not when they arrived; handle late data explicitly.
• Schema evolution: additive changes are safe; renames and type changes need versioning and coordination.
• Delivery semantics: at-least-once delivery plus idempotent writes is the usual practical route to effectively-once results.

Data quality and validation

Dimensions

• Completeness: required fields present; expected row counts.
• Validity: types, ranges, allowed categories, formats.
• Uniqueness: no duplicate keys or events.
• Consistency: values agree across tables (every order has a customer).
• Timeliness: data arrives within its freshness window.
• Accuracy: values match reality, usually checked by sampling and audits.

Checks and contracts

Validation should run at pipeline boundaries and fail loudly. Libraries such as Great Expectations, pandera and TensorFlow Data Validation let you declare expectations; the logic is simple enough to see in plain pandas:

import pandas as pd

def validate(df: pd.DataFrame, ref_null_rate: dict, max_null_drift=0.02):
    errors = []
    required = {"user_id", "event_time", "amount", "country"}
    missing = required - set(df.columns)
    if missing:
        return [f"missing columns: {sorted(missing)}"]
    if df["user_id"].isna().any():
        errors.append("user_id has nulls")
    if df.duplicated(["user_id", "event_time"]).any():
        errors.append("duplicate (user_id, event_time) rows")
    if not df["amount"].between(0, 50_000).all():
        errors.append("amount outside [0, 50000]")
    bad = set(df["country"].dropna()) - {"US", "CA", "MX", "GB", "DE"}
    if bad:
        errors.append(f"unexpected country codes: {sorted(bad)}")
    for col, ref in ref_null_rate.items():
        rate = df[col].isna().mean()
        if abs(rate - ref) > max_null_drift:
            errors.append(f"{col} null rate {rate:.3f} vs reference {ref:.3f}")
    return errors

df = pd.DataFrame({"user_id": [1, 2, 2], "event_time": ["t1", "t2", "t2"],
                   "amount": [20.0, -5.0, 13.0], "country": ["US", "XX", None]})
print(validate(df, ref_null_rate={"country": 0.01}))

Running it reports the duplicate row, the negative amount, the unknown country code and the null-rate jump (0.333 vs 0.01). Note the null-rate drift check: distribution checks against a reference catch silent upstream changes that schema checks miss.

A data contract is an explicit, versioned agreement between a data producer and its consumers about schema, semantics, quality thresholds and freshness, enforced in CI and at runtime. Contracts turn "the app team renamed a field and the model broke three weeks later" into a failed build on the day of the change.

Labeling workflows

Supervised learning needs labels, and label quality caps model quality. A professional labeling workflow:

1. Guidelines: a written spec with definitions, edge cases and examples for each label. Iterate on it.
2. Pilot: several annotators label the same small batch; measure agreement; revise guidelines where they disagree.
3. Production labeling with gold items (known answers) mixed in to monitor annotator accuracy, and a fraction of items labeled by two or more annotators to keep measuring agreement.
4. Adjudication: an expert resolves disagreements; disagreements also reveal ambiguous guidelines or genuinely ambiguous items.
5. Efficiency: active learning prioritizes items the model is most uncertain about; model-assisted pre-labeling (including LLM-generated draft labels) speeds annotators up but can bias them toward the suggestion, so audit a sample without pre-labels.
6. Ethics and operations: fair pay and working conditions, especially for distressing content; access control for sensitive data.

Cohen's kappa

Raw percent agreement overstates reliability because annotators agree some of the time by chance. Cohen's kappa (Cohen, 1960) corrects for chance agreement between two annotators:

where $p_o$ is the observed proportion of agreement and $p_e$ is the agreement expected by chance given each annotator's label frequencies:

with $p_{A,k}$ the fraction of items annotator A assigned to class $k$ (likewise B). $\kappa = 1$ is perfect agreement, $\kappa = 0$ is chance level, and negative values are worse than chance.

Worked example: two annotators labeling 100 messages as spam or ham

1. Observed agreement: $p_o = (40 + 45)/100 = 0.85$.
2. Label frequencies: A says spam 50%, ham 50%; B says spam 45%, ham 55%.
3. Chance agreement: $p_e = 0.50 \times 0.45 + 0.50 \times 0.55 = 0.225 + 0.275 = 0.50$.
4. Kappa: $\kappa = (0.85 - 0.50)/(1 - 0.50) = 0.35/0.50 = 0.70$.

85% raw agreement becomes $\kappa = 0.70$ once chance is removed. A commonly cited (and admittedly arbitrary) scale by Landis and Koch (1977) calls 0.61-0.80 "substantial" agreement. In practice, judge kappa against your task: for subjective labels like toxicity, 0.6-0.7 may be good; for factual labels, it signals unclear guidelines.

The same computation with scikit-learn:

from sklearn.metrics import cohen_kappa_score

a = ["spam"] * 50 + ["ham"] * 50
b = ["spam"] * 40 + ["ham"] * 10 + ["spam"] * 5 + ["ham"] * 45
print(round(cohen_kappa_score(a, b), 3))   # 0.7

Kappa is sensitive to class imbalance: with very rare positives, high raw agreement can coexist with low kappa. For more than two annotators use Fleiss' kappa; for missing ratings or ordinal/interval labels use Krippendorff's alpha; for ordinal scales, weighted kappa credits near-misses.

Feature stores

A feature store manages features as reusable, versioned assets with two faces:

• An offline store (warehouse or lake tables) with full history, used to build training sets.
• An online store (a low-latency key-value database) holding the latest feature values per entity, used at inference.

Definitions are written once and materialized to both, which reduces training-serving skew and lets teams share features. Examples include the open-source Feast and features built into commercial ML and data platforms. A feature store is worth its operational cost when multiple models share features or when real-time features are needed; a single batch model often does fine with well-managed warehouse tables.

Point-in-time correctness

The most important job of the offline store is point-in-time-correct training data: for each training example (an entity and a timestamp), join each feature's value as it was known at that time, never a later value. Joining the latest values leaks the future into training, producing models that look excellent offline and fail in production.

import pandas as pd

labels = pd.DataFrame({
    "user_id": [1, 1, 2],
    "ts": pd.to_datetime(["2026-03-01", "2026-03-10", "2026-03-05"]),
    "churned": [0, 1, 0]})
features = pd.DataFrame({
    "user_id": [1, 1, 1, 2],
    "ts": pd.to_datetime(["2026-02-20", "2026-03-05", "2026-03-12", "2026-03-01"]),
    "logins_30d": [12, 4, 0, 7]})

train = pd.merge_asof(labels.sort_values("ts"), features.sort_values("ts"),
                      on="ts", by="user_id", direction="backward")
print(train)

merge_asof with direction="backward" picks, for each label row, the latest feature row at or before its timestamp. User 1 on 2026-03-10 gets logins_30d = 4 (from 03-05), not the 0 recorded on 03-12, which would leak the post-churn inactivity. In production, also account for feature availability lag: if a feature lands two hours after the event, shift its timestamps by two hours before joining.

Data versioning and lineage

• Versioning: immutable snapshots of training sets (DVC, lakeFS, or table formats with time travel such as Delta Lake and Apache Iceberg), referenced by every training run (see the previous lesson).
• Lineage: a graph of which jobs read and wrote which datasets, column by column where possible. Open standards such as OpenLineage let orchestrators and engines emit lineage events to a common catalog. Lineage answers impact questions ("which models consume this column we want to drop?") and audit questions ("which sources fed model v17?").
• Retention and deletion: lineage is also what makes it feasible to honor deletion requests across derived datasets.

Privacy in pipelines

Detecting PII

Personally identifiable information (PII) includes direct identifiers (names, emails, phone numbers, government IDs, account numbers) and quasi-identifiers that identify people in combination (ZIP code, birth date, gender). Detection approaches:

• Pattern rules (regular expressions with checksums, such as the Luhn check for card numbers) for structured identifiers: precise but brittle.
• Named-entity recognition models for names, locations and organizations in free text.
• Frameworks combining both, such as Microsoft Presidio.

No detector is perfect; measure recall on a labeled sample of your own data.

Redaction strategies

• Masking/removal: replace with a placeholder like [EMAIL]. Simplest; destroys linkage.
• Pseudonymization: replace with a consistent token derived by a keyed hash (HMAC) so records can still be joined, while the key is stored separately with strict access. Under laws such as the EU GDPR, pseudonymized data remains personal data for anyone who can re-identify it, including the organization holding the key.
• Generalization: coarsen quasi-identifiers (age to a 10-year band, ZIP to its first three digits).
• Minimization: the best protection is not collecting or keeping fields the model does not need.

import hashlib, hmac, os, re

KEY = os.environ.get("PSEUDO_KEY", "dev-only-key").encode()  # load from a secret manager in production
EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
PHONE = re.compile(r"\+?\d[\d\s().-]{7,}\d")

def pseudonym(value: str) -> str:
    digest = hmac.new(KEY, value.lower().encode(), hashlib.sha256).hexdigest()
    return f"<EMAIL_{digest[:10]}>"

def redact(text: str) -> str:
    text = EMAIL.sub(lambda m: pseudonym(m.group()), text)
    return PHONE.sub("[PHONE]", text)

print(redact("Contact Ana at ana.r@example.com or +1 (214) 555-0199."))

Emails become consistent pseudonyms (the same address always maps to the same token, enabling joins), and phone numbers are masked. The name "Ana" is not caught, which illustrates why free text also needs an NER-based detector. Also scrub logs, prompts and model outputs: LLM applications often leak PII through these side channels.

Dataset documentation: datasheets

Gebru et al.'s "Datasheets for Datasets" proposes that every dataset ship with a document answering questions in seven areas:

1. Motivation: why and by whom was it created, and who funded it?
2. Composition: what do instances represent, how many, what is missing, does it contain sensitive or personal data?
3. Collection process: how was data acquired, over what time frame, with what consent?
4. Preprocessing, cleaning and labeling: what was done, and is the raw data available?
5. Uses: what has it been used for, and what uses should be avoided?
6. Distribution: how and under what license is it shared?
7. Maintenance: who maintains it, how are errors reported, how is it versioned?

Related formats include Data Statements for NLP and Data Cards. Treat the datasheet as a living artifact versioned with the data; regulators and customers increasingly ask for this information (see AI governance frameworks).

Common pitfalls

• Label leakage through time: joining latest feature values instead of point-in-time values.
• Two code paths for training and serving features, producing skew.
• Non-idempotent jobs that duplicate data on retries.
• Validating schema only, missing distribution shifts (null rates, category frequencies, units).
• Trusting percent agreement without chance correction; ignoring disagreement as a signal of ambiguous guidelines.
• Regex-only PII detection on free text; forgetting logs and prompts.
• Undocumented datasets whose provenance, consent and license are unknown when a customer or regulator asks.

In practice

• Start with a layered ELT layout, time-partitioned, idempotent jobs and validation at each boundary.
• Write data contracts for the sources your models depend on and enforce them in the producers' CI.
• Run a labeling pilot, report Cohen's kappa (or Krippendorff's alpha), and revise guidelines until agreement is acceptable for the task before scaling.
• Build training sets with point-in-time joins, whether or not you adopt a feature store.
• Minimize, detect and pseudonymize PII as early in the pipeline as possible, and test detector recall.
• Version every training snapshot, emit lineage, and ship a datasheet with each dataset.