Posted on
• Artificial Intelligence

Artificial Intelligence Data Processing with Python

Author
  • User
    linuxbash
    Posts by this author
    Posts by this author

Artificial Intelligence Data Processing with Python: A Linux Bash-first Playbook

Your model is only as good as your data pipeline. Most wins (and most failures) in AI happen before the first epoch: ingestion, cleaning, validation, feature engineering, and efficient storage. This post gives you a practical, Bash-first workflow for building reliable, fast Python data pipelines on Linux—so you can spend less time wrangling and more time modeling.

What you’ll get:

  • Why Python + Bash is a power combo for AI data prep

  • A minimal, reproducible toolkit you can install with apt, dnf, or zypper

  • 3–5 actionable steps with real-world examples and copy-paste code

  • A clear next step to productionize your pipeline

Why this matters (and why Python + Bash)

  • Reproducibility: Bash scripts + pinned Python environments make your pipeline deterministic and automatable (cron, systemd, CI).

  • Speed and cost: Efficient formats (Parquet), columnar processing, and streaming/chunking reduce memory and CPU.

  • Ecosystem leverage: Bash for orchestration and OS-level tooling; Python for rich data libraries (pandas, scikit-learn, pyarrow, polars).

0) Install the tooling

Install the base system packages. Choose your distro:

  • Debian/Ubuntu (apt):
sudo apt update
sudo apt install -y \
  python3 python3-venv python3-pip python3-dev build-essential \
  git curl jq parallel zstd pigz pkg-config libopenblas-dev
  • Fedora/RHEL/CentOS Stream (dnf):
sudo dnf install -y \
  python3 python3-pip python3-devel gcc gcc-c++ make \
  git curl jq parallel zstd pigz openblas-devel pkgconf-pkg-config
  • openSUSE (zypper):
sudo zypper refresh
sudo zypper install -y \
  python3 python3-pip python3-devel gcc gcc-c++ make \
  git curl jq zstd pigz openblas-devel pkgconf-pkg-config
# Note: GNU Parallel package may be named 'gnu_parallel' on some openSUSE versions:
# sudo zypper install -y parallel || sudo zypper install -y gnu_parallel

Create and populate a Python virtual environment:

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip wheel
pip install \
  numpy pandas pyarrow fastparquet polars \
  scikit-learn scipy pillow pandera datasets \
  csvkit

Verify:

python3 --version
python - <<'PY'
import numpy, pandas, pyarrow, sklearn, polars, pandera
print("Python data stack OK")
PY

Project skeleton:

mkdir -p data/{raw,staged,processed,cache} scripts logs

1) Ingest and validate your data (fast, safe, repeatable)

Pull data and sanity-check it before you touch a line of Python.

  • Download + decompress + quick peek:
URL="https://example.org/logs.csv.zst"
curl -L "$URL" -o data/raw/logs.csv.zst
zstd -d --force data/raw/logs.csv.zst -o data/raw/logs.csv
head -n 5 data/raw/logs.csv
sha256sum data/raw/logs.csv | tee data/raw/logs.csv.sha256
  • Quick stats with CLI tools (great for spot-checks in CI):
csvstat data/raw/logs.csv | head -n 40
jq --version  # keep handy for JSON lines streams
  • Schema validation with Python to catch problems early:
# scripts/validate.py
import pandas as pd
import pandera as pa
from pandera import Column, Check

schema = pa.DataFrameSchema({
    "ts": Column(pa.DateTime, coerce=True),
    "user_id": Column(pa.String, nullable=False),
    "latency_ms": Column(pa.Float, Check.ge(0)),
    "status": Column(pa.Int, Check.isin([200, 301, 302, 400, 401, 403, 404, 500])),
    "path": Column(pa.String)
})

df = pd.read_csv("data/raw/logs.csv")
schema.validate(df, lazy=True)
df.to_parquet("data/staged/logs.parquet", engine="pyarrow", index=False)
print("Validated and staged → data/staged/logs.parquet")

Run it:

source .venv/bin/activate
python scripts/validate.py

Why Parquet now? It’s columnar, compressed, and fast for downstream transforms and training.

2) Clean and transform to model-ready form

Normalize types, handle outliers/missing values, and add lightweight derived columns.

# scripts/clean_transform.py
import pandas as pd
import numpy as np

def load_and_fix(path: str) -> pd.DataFrame:
    df = pd.read_parquet(path)
    # Derived features
    df["hour"] = df["ts"].dt.floor("H")
    df["is_error"] = (df["status"] >= 400).astype(np.int8)
    # Clip and impute
    df["latency_ms"] = df["latency_ms"].clip(0, 10000)
    df = df.dropna(subset=["user_id", "path"])
    return df

df = load_and_fix("data/staged/logs.parquet")
df.to_parquet("data/processed/logs_clean.parquet", engine="pyarrow", index=False)
print("Cleaned → data/processed/logs_clean.parquet")

Large files? Stream in chunks:

# scripts/csv_to_parquet_chunked.py
import pandas as pd

out = "data/staged/logs.parquet"
writer = None
for chunk in pd.read_csv("data/raw/logs.csv", chunksize=500_000, parse_dates=["ts"]):
    if writer is None:
        chunk.to_parquet(out, engine="pyarrow", index=False)
    else:
        chunk.to_parquet(out, engine="pyarrow", index=False, append=True)  # requires pyarrow >= 14
    writer = True
print("Chunked conversion complete.")

Or go even faster with Polars:

# scripts/csv_to_parquet_polars.py
import polars as pl
(
  pl.scan_csv("data/raw/logs.csv", infer_schema_length=10000, try_parse_dates=True)
    .with_columns([
      pl.col("status").cast(pl.Int32),
      pl.col("latency_ms").cast(pl.Float64).clip_min(0)
    ])
    .collect()
    .write_parquet("data/staged/logs.parquet")
)

3) Feature engineering that scales

Tabular example (numerical + categorical) into sparse matrices:

# scripts/features_tabular.py
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.pipeline import Pipeline
from scipy import sparse
import numpy as np

df = pd.read_parquet("data/processed/logs_clean.parquet")

y = df["is_error"].to_numpy()
X_df = df[["latency_ms", "hour", "path", "user_id"]].copy()
X_df["hour"] = (pd.to_datetime(X_df["hour"]).astype("int64") // 10**9)

ct = ColumnTransformer(
    transformers=[
        ("num", StandardScaler(with_mean=True), ["latency_ms", "hour"]),
        ("cat", OneHotEncoder(handle_unknown="ignore", min_frequency=10), ["path", "user_id"]),
    ],
)

X = ct.fit_transform(X_df)
sparse.save_npz("data/processed/X.npz", X)
np.save("data/processed/y.npy", y)
print("Saved features to data/processed/X.npz and labels to data/processed/y.npy")

Text to TF-IDF (solid baseline, fast to train):

# scripts/text_to_tfidf.py
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from scipy import sparse

df = pd.read_csv("data/raw/reviews.csv")
tfidf = TfidfVectorizer(lowercase=True, max_features=50000, ngram_range=(1,2))
X = tfidf.fit_transform(df["review_text"].fillna(""))
sparse.save_npz("data/processed/reviews_tfidf.npz", X)
df["label"].to_csv("data/processed/reviews_labels.csv", index=False)
print("Text features ready.")

Images: standardize size and color space for CNNs:

# scripts/resize_images.py
from PIL import Image
from pathlib import Path

src = Path("data/raw/images")
dst = Path("data/processed/images_224")
dst.mkdir(parents=True, exist_ok=True)

for p in src.rglob("*.jpg"):
    im = Image.open(p).convert("RGB").resize((224, 224))
    out = dst / p.relative_to(src)
    out.parent.mkdir(parents=True, exist_ok=True)
    im.save(out, quality=95)
print("Resized images → data/processed/images_224")

4) Parallelize and cache for speed

Sharding + GNU Parallel multiplies throughput on multicore machines.

  • Convert multiple CSV shards to Parquet in parallel:
# scripts/convert_one.py
import sys, pandas as pd
src, dst = sys.argv[1], sys.argv[2]
pd.read_csv(src).to_parquet(dst, engine="pyarrow", index=False)
print(f"{src} → {dst}")

Run it:

ls data/raw/*.csv | parallel --bar 'python scripts/convert_one.py {} data/staged/{/.}.parquet'
  • Keep a manifest for reproducibility:
date -u +"%Y-%m-%dT%H:%M:%SZ" | tee data/processed/run_timestamp.txt
sha256sum data/processed/* | tee data/processed/manifest.sha256
  • Cache expensive steps (simple convention):
[ -f data/staged/logs.parquet ] || python scripts/csv_to_parquet_chunked.py
[ -f data/processed/logs_clean.parquet ] || python scripts/clean_transform.py

5) Real-world mini pipelines

  • Web log anomaly dataset:

    • Ingest logs → validate schema → clip latencies → error flag → one-hot path/user → save X.npz/y.npy.
    • Train a quick baseline (e.g., LogisticRegression) before investing in deep models.

  • Image classification:

    • rsync/curl the archive → verify sha256 → resize to 224×224 → split into train/val/test → write CSV manifest of file paths and labels.

  • Text sentiment:

    • Normalize quotes/whitespace → drop empties → TF-IDF with n-grams → sparse save → stratified split.

Optional: Makefile to glue it together

# Makefile (run with: make setup ingest validate transform features)
VENV=.venv
ACT=source $(VENV)/bin/activate

setup:
    python3 -m venv $(VENV)
    $(ACT) && pip install -U pip wheel
    $(ACT) && pip install numpy pandas pyarrow fastparquet polars scikit-learn scipy pillow pandera datasets csvkit

ingest:
    mkdir -p data/{raw,staged,processed,cache}
    curl -L $(URL) -o data/raw/data.csv.zst
    zstd -d --force data/raw/data.csv.zst -o data/raw/data.csv
    sha256sum data/raw/data.csv > data/raw/data.csv.sha256

validate:
    $(ACT) && python scripts/validate.py

transform:
    $(ACT) && python scripts/clean_transform.py

features:
    $(ACT) && python scripts/features_tabular.py

Conclusion and next steps (CTA)

Data pipelines win championships. With a Bash-first mindset and Python’s data stack, you can:

  • Ingest and validate deterministically

  • Transform at scale with columnar formats

  • Engineer features that are fast to train on

Your next step:

  • Pick one of the examples above and wire it end-to-end in your project. Start with Parquet + a simple feature script. Add parallelism once it works.

  • Productionize: add a Makefile target per step, schedule with cron/systemd, and log outputs to the logs/ directory.

  • Version your data artifacts (e.g., via checksums and dated folders) and pin your Python dependencies (requirements.txt).

If you want a quick win, copy the validate → transform → features scripts into your repo, adapt column names to your data, and run them today. Your future training runs will thank you.

Related posts