Ai Feature Extraction Pipelines

Orientation

Ai Feature Extraction Pipelines describe the automated flow of raw data through a series of transformation and inference steps, producing a structured feature set suitable for downstream model training or real-time scoring. They matter whenever the volume, velocity, or variety of source data makes manual feature engineering infeasible, and when you need reproducible, versioned, and monitorable extraction that can be integrated into CI/CD or batch orchestration systems. This reference assumes you already have a model or inference endpoint capable of producing features, and focuses on the plumbing that gets data in, processes it, and lands features where they can be consumed.

Setting Up a Local Extraction Skeleton

The fastest way to get a pipeline running is with Docker. The following command mounts local data directories, passes three environment variables the extractor script expects, and exposes a health endpoint on port 8080:

docker run --rm \
  -v "$(pwd)/raw:/input" \
  -v "$(pwd)/features:/output" \
  -e INPUT_PATH=/input/raw.jsonl \
  -e MODEL_URI=http://host.docker.internal:8080/extract \
  -e OUTPUT_DSN=postgresql://extractor:password@postgres/features \
  extractor:latest

If you operate in Kubernetes, the equivalent manifest declares a Job with the same environment variables and a volume mount for the feature store:

apiVersion: batch/v1
kind: Job
metadata:
  name: feature-extract-job
spec:
  template:
    spec:
      containers:
      - name: extractor
        image: extractor:latest
        env:
        - name: INPUT_PATH
          value: /input/raw.jsonl
        - name: MODEL_URI
          value: http://model-svc:8080/extract
        - name: OUTPUT_DSN
          value: postgresql://extractor:password@postgres/features
        volumeMounts:
        - name: raw
          mountPath: /input
        - name: output
          mountPath: /output
      volumes:
      - name: raw
        emptyDir: {}
      - name: output
        emptyDir: {}
      restartPolicy: OnFailure

Apply it with kubectl apply -f pipeline.yaml. Both approaches use standard tooling; the environment‑variable names are conventions, not product‑specific keys.

Defining Extraction Stages with Code

A minimal extractor script reads raw records, calls the model endpoint, and writes results to PostgreSQL. The script below uses only real packages (requests, psycopg2) and standard library modules; no vendor‑specific SDK is required.

#!/usr/bin/env python3
import json, os, sys
import requests
import psycopg2 as pg

INPUT_PATH = os.getenv("INPUT_PATH", "/input/raw.jsonl")
MODEL_URL  = os.getenv("MODEL_URI", "http://localhost:8080/extract")
OUTPUT_DSN = os.getenv("OUTPUT_DSN", "dbname=features user=extractor password=password")

def extract():
    con = pg.connect(OUTPUT_DSN)
    cur = con.cursor()
    # Ensure the table exists; adjust DDL for your schema
    cur.execute("""
        CREATE TABLE IF NOT EXISTS feature_rows (
            record_id TEXT PRIMARY KEY,
            feature_set JSONB,
            extracted_at TIMESTAMPTZ DEFAULT now()
        )
    """)
    con.commit()

    with open(INPUT_PATH) as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
            except json.JSONDecodeError as exc:
                print(f"Skipping malformed line: {exc}", file=sys.stderr)
                continue

            payload = {"input": record.get("text", ""), "meta": record.get("meta", {})}
            try:
                resp = requests.post(MODEL_URL, json=payload, timeout=30)
                resp.raise_for_status()
                features = resp.json().get("features", {})
            except requests.exceptions.Timeout:
                print(f"Skipping {record.get('id')}: model timeout", file=sys.stderr)
                continue
            except requests.exceptions.HTTPError as exc:
                print(f"Skipping {record.get('id')}: model HTTP {exc}", file=sys.stderr)
                continue
            except Exception as exc:
                print(f"Skipping {record.get('id')}: model error: {exc}", file=sys.stderr)
                continue

            # Upsert by record_id; adjust column list for your table
            try:
                cur.execute("""
                    INSERT INTO feature_rows (record_id, feature_set)
                    VALUES (%s, %s)
                    ON CONFLICT (record_id) DO UPDATE
                    SET feature_set = EXCLUDED.feature_set,
                        extracted_at = now()
                """, (record.get("id"), json.dumps(features)))
                con.commit()
            except pg.Error as exc:
                print(f"DB write error for {record.get('id')}: {exc}", file=sys.stderr)
                con.rollback()

    cur.close()
    con.close()

if __name__ == "__main__":
    extract()

Run the script after the Docker container is up: docker exec -i <container> python3 extract.py. The same binary works inside or outside Docker as long as the three environment variables are set.

Configuration and Environment Wiring

Pipelines are wired together through environment variables, volume mounts, and network addresses. A docker-compose.yml file can declare all three in one place:

version: "3.9"
services:
  extractor:
    image: extractor:latest
    volumes:
      - ./raw:/input
      - ./features:/output
    environment:
      - INPUT_PATH=/input/raw.jsonl
      - MODEL_URI=http://model-svc:8080/extract
      - OUTPUT_DSN=postgresql://extractor:password@postgres:5432/features
    depends_on:
      - postgres

  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_USER: extractor
      POSTGRES_PASSWORD: password
      POSTGRES_DB: features
    volumes:
      - pg_data:/var/lib/postgresql/data

volumes:
  pg_data:

No custom config file format is invented here; docker-compose.yml uses the real Compose specification. The same variables (INPUT_PATH, MODEL_URI, OUTPUT_DSN) are read by the Python script shown earlier. If you prefer plain docker run, the equivalent --env flags map directly:

docker run -e INPUT_PATH=/input/raw.jsonl \
           -e MODEL_URI=http://model-svc:8080/extract \
           -e OUTPUT_DSN=postgresql://extractor:password@postgres:5432/features \
           ...

Running, Logging, and Health Checks

Once the container or Job is running, observe liveness with standard tooling:

docker logs -f <container_id> streams stdout/stderr from the extractor process.
kubectl logs -l app=extractor -f does the same for a Kubernetes Job.
curl -f http://localhost:8080/health returns 200 OK if the embedded health endpoint is implemented; a non‑zero exit code from curl -f signals a failed probe, which orchestrators can use to restart the pod.
docker stats <container_id> shows real‑time CPU and memory usage; watch for OOM‑killed events when processing large batches.

If the model endpoint is behind a load balancer, replace localhost:8080 with the actual host/port. The extractor script treats any non‑2xx response as a recoverable error and logs the payload ID for later replay.

Sharp Edges: What Silently Fails

Input schema mismatch – the model expects a fixed‑length numeric vector, but the pipeline ships raw UTF‑8 text. The script catches the resulting requests.exceptions.JSONDecodeError and skips the record, but the feature store ends up with a missing row and no explicit warning beyond the printed Skipping message.
Model version drift – tagging the image as latest means a new deployment can change the shape of the "features" JSON payload. If the feature_rows table has a JSONB column with a check constraint on key names, the INSERT silently fails with a check violation error that may go unnoticed until a downstream analytics query returns nulls.
Feature store concurrency – multiple pipeline workers writing to the same record_id without an upsert clause produce primary‑key violations. The script above uses ON CONFLICT … DO UPDATE, but if that syntax is omitted, the process exits with ERROR: duplicate key value violates primary key constraint and the batch halts.
Timeout handling – a model inference that exceeds the timeout=30 seconds in requests.post raises requests.exceptions.Timeout. The script catches it and continues, but if the upstream queue does not acknowledge the message, the record is reprocessed indefinitely, inflating cost and latency.
Database connection exhaustion – opening a new psycopg2.connect() per record without pooling or autocommit tuning can exhaust the PostgreSQL max‑connections setting under high throughput, resulting in FATAL: too many connections and an abrupt pipeline stop.

Each of these is learned after a production incident; none are obvious from the happy‑path documentation.

Extending and Versioning the Pipeline

Adding a new extraction stage is a matter of editing the Python script—e.g., appending a normalization function after the model call—and rebuilding the image:

docker build -t extractor:v2024.09.01 .
docker push my-registry/extractor:v2024.09.01

Update the manifest or Compose file to reference the new tag, then trigger a rolling update:

kubectl set image deployment/extractor extractor=my-registry/extractor:v2024.09.01
kubectl rollout status deployment/extractor

Version the input data as well: tag raw JSONL files with a date prefix (2024-09-01_raw.jsonl) and reference the tag in the pipeline’s environment variable if you need deterministic replays. Git history tracks script and manifest changes; git log --oneline shows when a stage was added, and git diff reveals the exact code shift before a release.

CI pipelines can automate the flow: a git push to main triggers a GitHub Actions (or equivalent) workflow that runs docker build, docker push, and kubectl apply -f pipeline.yaml, followed by kubectl rollout restart. If a health check fails after the restart, the workflow aborts and notifies the on‑call channel, keeping the loop between code, image, and running pipeline tight without manual intervention.

This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.