Introduction
Artificial intelligence and machine learning workloads demand robust, scalable storage infrastructure that can handle massive datasets, frequent model checkpoints, and real-time inference requirements. Modern AI applications generate and consume data at unprecedented scales - training datasets can reach petabytes, model artifacts require versioning and lineage tracking, and inference systems need millisecond access to features and embeddings.
MinIO emerges as the ideal storage foundation for AI workloads, offering S3-compatible object storage with enterprise-grade performance, security, and scalability. Unlike traditional file systems that struggle with AI’s unique access patterns, MinIO provides high-throughput parallel I/O, seamless cloud integration, and native support for modern data formats like Parquet and Delta Lake.
Key advantages of MinIO for AI include:
- High Performance: Multi-part uploads and parallel processing for large datasets
- S3 Compatibility: Works with existing ML frameworks and tools
- Kubernetes Native: Cloud-native deployment and auto-scaling
- Data Governance: Versioning, encryption, and access controls
- Cost Efficiency: Intelligent tiering and lifecycle management
This comprehensive guide explores building production AI infrastructure with MinIO, covering everything from data lake architecture to real-time inference pipelines.
ML Data Lake Architecture
Designing the Data Lake Foundation
A well-architected ML data lake on MinIO follows a layered approach that separates raw ingestion, processed data, feature engineering, and model artifacts. This separation enables different teams to work independently while maintaining data lineage and governance.
The MLDataLake class below demonstrates how to bootstrap this foundation with a single S3 client and a fixed set
of purpose-built buckets. Each bucket maps to a distinct stage of the data lifecycle: raw-data holds immutable
source files exactly as they arrived, bronze-data stores cleaned and validated versions, silver-data contains
enriched and feature-ready data, and gold-data exposes business-ready aggregates and training splits. The
remaining buckets - feature-store, model-registry, experiment-tracking, and inference-cache - are
operational outputs of the ML lifecycle rather than raw storage layers.
Several design decisions in the code are worth calling out. First, the client is configured with
max_pool_connections=100 and a retry budget of three attempts, which prepares the connection pool for the bursty,
parallel access patterns that dominate training workloads. Second, every bucket receives tags describing its
purpose and environment, turning MinIO into a self-documenting system where S3 tagging serves as lightweight
metadata for governance tooling. Third, setup_buckets() is idempotent - it tolerates buckets that already exist,
so the class can be invoked safely at startup by multiple workers or in test environments.
This bucket-per-stage model is the storage equivalent of the Medallion architecture popularized by Databricks, and it scales from a single notebook to a multi-team data platform. The main trade-off is that objects are spread across many buckets, which complicates cross-stage queries; teams that prefer a single-bucket layout can achieve the same separation with top-level prefixes instead. The rest of this guide builds directly on these buckets, so their names recur throughout the ingestion, feature store, and registry code that follows.
import boto3
import json
from datetime import datetime
from botocore.config import Config
class MLDataLake:
def __init__(self, endpoint_url, access_key, secret_key):
self.s3 = boto3.client(
's3',
endpoint_url=endpoint_url,
aws_access_key_id=access_key,
aws_secret_access_key=secret_key,
config=Config(
max_pool_connections=100,
retries={'max_attempts': 3}
)
)
self.setup_buckets()
def setup_buckets(self):
"""Initialize data lake bucket structure"""
buckets = {
'raw-data': 'Raw ingested data from various sources',
'bronze-data': 'Cleaned and validated data',
'silver-data': 'Transformed and enriched data',
'gold-data': 'Business-ready aggregated data',
'feature-store': 'Computed features for ML',
'model-registry': 'Trained models and metadata',
'experiment-tracking': 'ML experiment artifacts',
'inference-cache': 'Cached predictions and embeddings'
}
for bucket, description in buckets.items():
try:
self.s3.create_bucket(Bucket=bucket)
# Add bucket metadata
self.s3.put_bucket_tagging(
Bucket=bucket,
Tagging={
'TagSet': [
{'Key': 'Purpose', 'Value': 'ML-DataLake'},
{'Key': 'Description', 'Value': description},
{'Key': 'Environment', 'Value': 'production'}
]
}
)
print(f"Created bucket: {bucket}")
except Exception as e:
print(f"Bucket {bucket} already exists or error: {e}")
# Initialize data lake
data_lake = MLDataLake(
endpoint_url='http://minio:9000',
access_key='minioadmin',
secret_key='minioadmin'
)
Walking through the class, __init__() does two things in sequence: it builds the low-level S3 client and then
immediately calls setup_buckets(). Building the client with Config rather than relying on defaults is a
deliberate choice, because the stock boto3 settings are tuned for latency-sensitive single-object calls rather than
the throughput-heavy access patterns of AI workloads. A connection pool of 100 sockets means that many parallel
uploads and downloads can proceed without queueing, and the three-attempt retry budget absorbs transient network
failures while still failing fast when MinIO is genuinely unavailable. Notice that the client is created once and
shared across all bucket operations; creating a new client per request would silently destroy the connection-pool
benefit.
The setup_buckets() loop is the heart of the pattern. It iterates over a dictionary whose values are
human-readable descriptions, creates each bucket, and then tags it with Purpose, Description, and
Environment. Those tags are not decorative: S3 tagging is queryable through the API, appears in MinIO’s web
console, and can be used by lifecycle policies, replication rules, and budget reports to act on groups of buckets
without hard-coding names. Encoding the environment as a tag rather than in the bucket name is a small but
important decision - it lets the same code create identical bucket names in dev, staging, and production while the
tags distinguish them, which keeps data paths consistent across environments.
The exception handling around create_bucket is another quiet but meaningful detail. In a fresh deployment the
loop succeeds the first time, but on restart - or when a test runner calls the constructor repeatedly - the buckets
already exist and create_bucket raises BucketAlreadyOwnedByYou. The bare except block converts that into a
harmless log line, making the whole initializer idempotent. That idempotence matters more than it first appears,
because data-lake bootstrap code often runs from scheduled jobs, from multiple notebook kernels, or from a cluster
of workers, all of which may race to create the same buckets.
One more point worth internalizing: nothing in this class moves data. Its entire job is to establish a contract - a
set of named, tagged destinations that every later pipeline will write to and read from. The engineering value is
in that stability: once feature-store code targets feature-store/features/..., it never needs to know which
server, disk, or volume backs it, because MinIO abstracts all of that away. That is precisely why the class is
worth keeping even after buckets are provisioned by Terraform: the constructor doubles as a self-healing bootstrap
that guarantees the contract exists no matter how the process was started.
Data Organization Strategy
Structuring objects with meaningful, partitioned prefixes is one of the most consequential decisions in a data
lake, and it is far easier to design up front than to retrofit later. The create_data_structure() function below
encodes a Hive-style partitioning convention in which each path segment is a key=value pair - year=2026,
source=web, sample_rate=44100 - that acts as a column in downstream query engines. Partitioning on time-based
or high-cardinality columns lets Spark, Trino, and DuckDB prune entire directories without scanning them, which
cuts query cost dramatically as datasets grow into the terabyte range.
The example also preserves the data type of each stream: images keep their resolution and capture date, audio records the format and sample rate, text tracks its source and language, and structured tables record their schema version. This self-describing layout means that a scientist can find and understand any dataset by reading the object path alone, without consulting a separate catalog. It is a lightweight alternative to a full metadata layer such as a Hive metastore or Unity Catalog, and it is often sufficient for teams of one to fifty people.
The second function, store_partitioned_data(), shows how a pandas DataFrame becomes partitioned Parquet objects
on MinIO. Grouping by the partition columns, constructing the key=value path, and writing a single data.parquet
per group produces exactly the layout that the earlier structure declares. Parquet is chosen deliberately: its
columnar format compresses well, is splittable for parallel reads, and is supported natively by every major engine.
The main trade-off is that small files - one per group - hurt performance, so batch by time window or partition
cardinality rather than writing millions of tiny objects, and consider coalescing partitions before the write.
def create_data_structure():
"""Create standardized directory structure for ML projects"""
structure = {
'raw-data': [
'images/year=2026/month=01/day=01/',
'text/source=web/year=2026/month=01/',
'audio/format=wav/sample_rate=44100/',
'video/resolution=1080p/fps=30/',
'structured/format=parquet/schema_version=v1/'
],
'bronze-data': [
'images/processed/year=2026/month=01/',
'text/cleaned/language=en/',
'audio/normalized/duration_sec=30/',
'video/frames/extracted_fps=1/',
'structured/validated/schema_version=v1/'
],
'silver-data': [
'features/image_embeddings/model=resnet50/',
'features/text_embeddings/model=bert/',
'features/audio_mfcc/window_size=25ms/',
'aggregations/daily/metric=engagement/',
'joins/user_content/date=2026-01-01/'
],
'gold-data': [
'datasets/training/split=train/version=v1/',
'datasets/validation/split=val/version=v1/',
'datasets/test/split=test/version=v1/',
'reports/model_performance/date=2026-01-01/',
'dashboards/business_metrics/refresh=daily/'
]
}
return structure
# Example: Partitioned data storage
def store_partitioned_data(df, bucket, base_path, partition_cols):
"""Store DataFrame with partitioning for efficient querying"""
import pandas as pd
for partition_values, group in df.groupby(partition_cols):
# Create partition path
partition_path = base_path
for col, value in zip(partition_cols, partition_values):
partition_path += f"/{col}={value}"
# Save as Parquet
parquet_buffer = group.to_parquet(index=False)
key = f"{partition_path}/data.parquet"
s3.put_object(
Bucket=bucket,
Key=key,
Body=parquet_buffer,
ContentType='application/octet-stream'
)
Reading the structure dictionary more closely, it is easy to see the mental model the lake is built around. The
raw-data layer is immutable: objects arrive with whatever naming the source produced and are never edited in
place, which makes raw ingestion cheap, reproducible, and auditable. The bronze-data and silver-data layers
progressively add value - cleaning, normalization, embeddings, and joins - while each prefix records the exact
processing parameters (model name, window size, schema version) that produced it. That last detail is subtle but
crucial: a prefix like features/audio_mfcc/window_size=25ms says more about a dataset than any README, because it
captures the configuration that determines the data’s semantics.
The store_partitioned_data() function also illustrates a critical engineering point about how MinIO behaves under
the hood. Writing one data.parquet per partition means that reads can be restricted to a single object per
partition group; a query engine opening only date=2026-01-01 touches one small file instead of scanning
everything. However, the naive loop shown here writes one object per group, and if partitions are very fine-grained
the lake fills with tiny files that degrade read throughput. In a real deployment you would call this function with
coarse partition columns such as year and month, or batch groups together, so that each Parquet object lands in
a healthy size range of tens to hundreds of megabytes.
There is also a meaningful choice in the ContentType of these objects. Because MinIO stores object metadata
alongside the data, setting ContentType='application/octet-stream' for Parquet prevents MinIO from attempting any
interpretation of the bytes; the schema lives inside the Parquet footer, not in HTTP headers. For web-facing or
dashboard data you would instead set a proper MIME type so that browsers or CDNs handle the object correctly. The
general lesson is that MinIO objects are just bytes plus a metadata envelope, and the envelope should be populated
with whatever makes downstream tooling - query engines, browsers, or your own code - most productive.
Data Governance and Lineage
As pipelines multiply, the ability to explain how a given dataset was produced becomes a compliance requirement as
much as an engineering nicety. The DataLineageTracker below writes a JSON record for every transformation,
capturing input paths, the output path, a hash of the transformation code, the parameters used, and execution
metrics such as duration, memory, and CPU. Because MinIO is S3-compatible, these records can be stored right next
to the data in a dedicated data-lineage bucket, which keeps lineage close to the artifacts it describes and makes
it trivially auditable by any S3 tooling.
The lineage key uses the output path and a timestamp, guaranteeing uniqueness and enabling a chronological history of every version of a dataset. Storing execution metadata (duration, memory, CPU cores) in the same record turns lineage into a cost and performance audit trail: you can later answer questions like “which transformation consumed the most resources” or “which code change produced this anomaly” simply by reading the JSON files.
The deliberate simplicity of this design is its strength. Production systems often integrate with full-featured
tools such as OpenLineage or DataHub, but a JSON-per-transformation approach is immediately implementable, requires
no new infrastructure, and remains fully queryable via list_objects_v2 and get_object. The main limitation is
the lack of a query engine over the lineage records; teams that outgrow this pattern can load the bucket into
DuckDB or Athena with a few lines of SQL.
class DataLineageTracker:
def __init__(self, s3_client):
self.s3 = s3_client
self.lineage_bucket = 'data-lineage'
def track_transformation(self, input_paths, output_path,
transformation_code, metadata):
"""Track data transformation lineage"""
lineage_record = {
'timestamp': datetime.now().isoformat(),
'input_paths': input_paths,
'output_path': output_path,
'transformation': {
'code_hash': hash(transformation_code),
'description': metadata.get('description', ''),
'parameters': metadata.get('parameters', {}),
'framework': metadata.get('framework', 'python')
},
'execution_info': {
'duration_seconds': metadata.get('duration', 0),
'memory_usage_mb': metadata.get('memory_usage', 0),
'cpu_cores': metadata.get('cpu_cores', 1)
}
}
# Store lineage record
lineage_key = f"lineage/{output_path.replace('/', '_')}/{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
self.s3.put_object(
Bucket=self.lineage_bucket,
Key=lineage_key,
Body=json.dumps(lineage_record, indent=2),
ContentType='application/json'
)
return lineage_record
# Usage example
tracker = DataLineageTracker(s3)
tracker.track_transformation(
input_paths=['raw-data/images/2026/01/01/'],
output_path='bronze-data/images/processed/2026/01/01/',
transformation_code="resize_and_normalize()",
metadata={
'description': 'Resize images to 224x224 and normalize pixel values',
'parameters': {'target_size': (224, 224), 'normalize': True},
'duration': 120.5,
'memory_usage': 2048
}
)
Look at what track_transformation() actually persists, because each field earns its place. The code_hash is
computed with Python’s built-in hash(), which is fast but randomized per process for strings; in a long-running
system you would swap it for a stable sha256 so that the same code always maps to the same fingerprint. The
transformation block records the framework and parameters, and the execution_info block captures resource
usage. Together these enable a very practical workflow: when a model’s behavior drifts, you can trace which dataset
version, which parameters, and which code produced the training data, and then either re-run the transformation or
reproduce the failure locally.
The key format deserves attention too. By encoding the output path (with slashes flattened) and a timestamp into the object name, the tracker guarantees that every run produces a unique object and that a full chronological history accumulates automatically. That means you never overwrite the past - old lineage records remain readable forever, which is exactly what auditors and data scientists want. If the same transformation runs twice a second, the timestamp collision risk is real, so production versions should append a monotonic sequence or a UUID rather than relying on second-level timestamps alone.
This pattern slots naturally into the pipelines shown elsewhere in the guide. The HighThroughputIngestion class
can call the tracker after each batch upload, the Spark pipeline can emit a record when it finishes writing a
parquet split, and the feature store can log every feature group registration. Because the tracker only needs an S3
client, it imposes no framework coupling, and it can be composed into orchestrators like Airflow or Prefect as a
callback. The cost is a few extra objects per transformation - negligible in exchange for a complete audit trail.
Training Data Pipelines with MinIO
High-Performance Data Ingestion
Modern ML training requires efficient data ingestion that can saturate GPU utilization. MinIO’s multi-part upload and parallel processing capabilities enable high-throughput data pipelines that scale with your training infrastructure.
import boto3
import os
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from botocore.config import Config
import hashlib
import json
class HighThroughputIngestion:
def __init__(self, endpoint_url, access_key, secret_key, max_workers=20):
self.s3 = boto3.client(
's3',
endpoint_url=endpoint_url,
aws_access_key_id=access_key,
aws_secret_access_key=secret_key,
config=Config(
max_pool_connections=max_workers,
retries={'max_attempts': 3}
)
)
self.max_workers = max_workers
self.upload_stats = {'success': 0, 'failed': 0, 'bytes': 0}
def upload_with_metadata(self, local_path, bucket, s3_key, metadata=None):
"""Upload file with comprehensive metadata"""
try:
# Calculate file hash for integrity
with open(local_path, 'rb') as f:
file_hash = hashlib.sha256(f.read()).hexdigest()
# Prepare metadata
upload_metadata = {
'sha256': file_hash,
'original_name': os.path.basename(local_path),
'upload_timestamp': str(int(time.time())),
'file_size': str(os.path.getsize(local_path))
}
if metadata:
upload_metadata.update(metadata)
# Upload with metadata
self.s3.upload_file(
local_path, bucket, s3_key,
ExtraArgs={
'Metadata': upload_metadata,
'ContentType': self._get_content_type(local_path)
}
)
self.upload_stats['success'] += 1
self.upload_stats['bytes'] += os.path.getsize(local_path)
return {'status': 'success', 'key': s3_key, 'hash': file_hash}
except Exception as e:
self.upload_stats['failed'] += 1
return {'status': 'failed', 'key': s3_key, 'error': str(e)}
def batch_upload_dataset(self, dataset_path, bucket, prefix,
file_extensions=None, metadata_extractor=None):
"""Upload entire dataset with parallel processing"""
if file_extensions is None:
file_extensions = ['.jpg', '.png', '.txt', '.parquet', '.csv']
# Collect all files to upload
upload_tasks = []
for root, dirs, files in os.walk(dataset_path):
for file in files:
if any(file.lower().endswith(ext) for ext in file_extensions):
local_path = os.path.join(root, file)
relative_path = os.path.relpath(local_path, dataset_path)
s3_key = f"{prefix}/{relative_path}"
# Extract metadata if function provided
metadata = {}
if metadata_extractor:
metadata = metadata_extractor(local_path)
upload_tasks.append((local_path, bucket, s3_key, metadata))
print(f"Starting upload of {len(upload_tasks)} files...")
# Execute parallel uploads
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
future_to_task = {
executor.submit(self.upload_with_metadata, *task): task
for task in upload_tasks
}
results = []
for future in as_completed(future_to_task):
result = future.result()
results.append(result)
if len(results) % 100 == 0:
print(f"Uploaded {len(results)}/{len(upload_tasks)} files")
return results
def _get_content_type(self, file_path):
"""Determine content type based on file extension"""
ext = os.path.splitext(file_path)[1].lower()
content_types = {
'.jpg': 'image/jpeg', '.jpeg': 'image/jpeg',
'.png': 'image/png', '.gif': 'image/gif',
'.txt': 'text/plain', '.csv': 'text/csv',
'.json': 'application/json',
'.parquet': 'application/octet-stream',
'.pt': 'application/octet-stream',
'.pkl': 'application/octet-stream'
}
return content_types.get(ext, 'application/octet-stream')
# Example: Upload ImageNet-style dataset
def extract_image_metadata(image_path):
"""Extract metadata from image files"""
from PIL import Image
try:
with Image.open(image_path) as img:
return {
'width': str(img.width),
'height': str(img.height),
'format': img.format,
'mode': img.mode,
'class_label': os.path.basename(os.path.dirname(image_path))
}
except Exception:
return {}
# Usage
ingestion = HighThroughputIngestion(
'http://minio:9000', 'minioadmin', 'minioadmin'
)
results = ingestion.batch_upload_dataset(
dataset_path='./imagenet_subset',
bucket='training-data',
prefix='imagenet/train',
file_extensions=['.jpg', '.jpeg'],
metadata_extractor=extract_image_metadata
)
print(f"Upload complete: {ingestion.upload_stats}")
The ingestion layer is deliberately built around parallelism and metadata. Tagging every object with a SHA-256 checksum at write time means corruption can be detected during later reads without re-hashing the original source, and the extracted image metadata (width, height, format, class label) becomes searchable without opening the files. If you upload a petabyte-scale corpus, switch from per-file threading to MinIO’s client-side multipart uploads with higher concurrency, and always run a verification pass that compares ETags against the computed hashes.
Data Validation and Quality Checks
Garbage-in, garbage-out is the oldest rule in machine learning, and validation is the gate that keeps bad data out
of the training loop. The DataQualityValidator class below downloads each file from MinIO and checks it against a
declarative set of rules - required columns, column types, null checks, and value ranges for tabular data, plus
dimensions, format, and size limits for images. Because validation runs against the objects stored in the lake
rather than in-memory copies, it catches corruption or drift in exactly the data that training will consume.
The rules dictionary is the interesting design choice: validation logic is data, not code. Defining rules like
'no_nulls': ['image_path', 'label'] or 'min_dimensions': (224, 224) as a plain structure means they can be
versioned, shared across teams, and even stored back in MinIO as a schema registry. It also keeps the validator
generic - the same class validates tabular Parquet, CSV, and image datasets by dispatching on file extension.
This step belongs between ingestion and any Spark transformation. Catching a truncated JPEG or a mislabeled column before processing saves hours of debugging later, because errors are attributed to the source data rather than to the pipeline that consumed it. The main trade-off is the cost of downloading every file; for very large lakes, sample-based validation or cheap object-metadata checks (size, ETag) can run first, with full validation reserved for files that pass those gates.
import pandas as pd
import numpy as np
from typing import Dict, List, Any
class DataQualityValidator:
def __init__(self, s3_client):
self.s3 = s3_client
self.validation_results = []
def validate_dataset(self, bucket, prefix, validation_rules):
"""Validate dataset quality across multiple files"""
# List all files in dataset
response = self.s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
validation_summary = {
'total_files': 0,
'valid_files': 0,
'invalid_files': 0,
'validation_errors': []
}
for obj in response.get('Contents', []):
key = obj['Key']
validation_summary['total_files'] += 1
try:
# Download and validate file
local_path = f"/tmp/{os.path.basename(key)}"
self.s3.download_file(bucket, key, local_path)
is_valid, errors = self._validate_file(local_path, validation_rules)
if is_valid:
validation_summary['valid_files'] += 1
else:
validation_summary['invalid_files'] += 1
validation_summary['validation_errors'].extend([
{'file': key, 'errors': errors}
])
# Cleanup
os.remove(local_path)
except Exception as e:
validation_summary['invalid_files'] += 1
validation_summary['validation_errors'].append({
'file': key, 'errors': [f"Processing error: {str(e)}"]
})
return validation_summary
def _validate_file(self, file_path, rules):
"""Validate individual file against rules"""
errors = []
try:
if file_path.endswith('.parquet'):
df = pd.read_parquet(file_path)
errors.extend(self._validate_dataframe(df, rules))
elif file_path.endswith(('.jpg', '.png')):
errors.extend(self._validate_image(file_path, rules))
elif file_path.endswith('.csv'):
df = pd.read_csv(file_path)
errors.extend(self._validate_dataframe(df, rules))
except Exception as e:
errors.append(f"File format error: {str(e)}")
return len(errors) == 0, errors
def _validate_dataframe(self, df, rules):
"""Validate DataFrame against rules"""
errors = []
# Check required columns
if 'required_columns' in rules:
missing_cols = set(rules['required_columns']) - set(df.columns)
if missing_cols:
errors.append(f"Missing columns: {missing_cols}")
# Check data types
if 'column_types' in rules:
for col, expected_type in rules['column_types'].items():
if col in df.columns and df[col].dtype != expected_type:
errors.append(f"Column {col} has type {df[col].dtype}, expected {expected_type}")
# Check for null values
if 'no_nulls' in rules:
for col in rules['no_nulls']:
if col in df.columns and df[col].isnull().any():
errors.append(f"Column {col} contains null values")
# Check value ranges
if 'value_ranges' in rules:
for col, (min_val, max_val) in rules['value_ranges'].items():
if col in df.columns:
if df[col].min() < min_val or df[col].max() > max_val:
errors.append(f"Column {col} values outside range [{min_val}, {max_val}]")
return errors
def _validate_image(self, image_path, rules):
"""Validate image file against rules"""
errors = []
try:
from PIL import Image
with Image.open(image_path) as img:
# Check dimensions
if 'min_dimensions' in rules:
min_w, min_h = rules['min_dimensions']
if img.width < min_w or img.height < min_h:
errors.append(f"Image too small: {img.width}x{img.height}")
# Check format
if 'allowed_formats' in rules:
if img.format not in rules['allowed_formats']:
errors.append(f"Invalid format: {img.format}")
# Check file size
if 'max_file_size_mb' in rules:
file_size_mb = os.path.getsize(image_path) / (1024 * 1024)
if file_size_mb > rules['max_file_size_mb']:
errors.append(f"File too large: {file_size_mb:.2f}MB")
except Exception as e:
errors.append(f"Image validation error: {str(e)}")
return errors
# Example validation rules
validation_rules = {
'required_columns': ['image_path', 'label', 'split'],
'column_types': {'label': 'int64', 'split': 'object'},
'no_nulls': ['image_path', 'label'],
'value_ranges': {'label': (0, 999)},
'min_dimensions': (224, 224),
'allowed_formats': ['JPEG', 'PNG'],
'max_file_size_mb': 10
}
# Run validation
validator = DataQualityValidator(s3)
results = validator.validate_dataset(
bucket='training-data',
prefix='imagenet/train',
validation_rules=validation_rules
)
print(f"Validation Results: {results}")
Validation rules written as data compose well with the data lake itself: store the rules dictionary in a schemas/
prefix, and have the validator load it from MinIO so that data engineers can change quality gates without
redeploying code. Whenever a validation run flags files, route them to a quarantine/ prefix instead of deleting
them, which preserves the ability to debug or recover while keeping bad data out of training.
Looking at the implementation details, the class has a clean separation of responsibilities. The outer
validate_dataset() method handles the mechanics of interacting with MinIO - listing objects under a prefix,
downloading each file to a temporary location, tracking counts, and cleaning up afterward - while the private
_validate_file() and rule-checking helpers operate purely on local files. This split means the storage layer and
the validation logic can evolve independently: you could swap boto3 for the S3-compatible API of another provider,
or add new rule types, without touching the other half.
The temporary-file pattern deserves a moment of attention. Downloading to /tmp and calling os.remove()
afterwards is pragmatic but has real consequences in a distributed setting. If the validator runs on a pod with a
small ephemeral disk, a large dataset can fill the volume; if a crash happens between download and cleanup,
orphaned temp files accumulate. Production validators often work around this by reading objects into memory with
get_object() for small files, using streaming reads for large ones, or mounting an ephemeral volume and wrapping
the whole run in a try/finally that empties it.
The rule engine itself is worth studying as a reference design. Each rule type - required_columns,
column_types, no_nulls, value_ranges, min_dimensions, allowed_formats, max_file_size_mb - is a small,
independently testable check that appends human-readable error strings to a list. Because rules are keyed by name
and applied conditionally, the same validator serves Parquet tabular data and JPEG image data with a single code
path. The error strings are structured for humans (Column label has type object, expected int64), but they could
just as easily be emitted as structured JSON for dashboards or alerting, since the validator returns a summary with
per-file error lists.
One practical gap is worth noting: validate_dataset() lists objects with a single list_objects_v2 call, which
returns at most 1,000 keys per request. For any realistic image dataset the listing will be truncated, so
production code must page through with a paginator or follow the NextContinuationToken. This is a recurring theme
across the guide - every list call against MinIO needs pagination handling once the bucket grows past a thousand
objects - and it is easy to miss during local testing with small directories.
Distributed Data Processing with Apache Spark
For data that exceeds a single machine’s memory, Apache Spark is the standard distributed processing engine, and it
reads and writes MinIO through the Hadoop S3A filesystem connector. The create_spark_session() function below
captures the configuration keys that make this work: the endpoint URL, the access and secret keys, path-style
access (essential for S3-compatible stores like MinIO that do not use virtual-hosted buckets), and the S3A
implementation class. Without path.style.access=true, Spark tries to reach the bucket as a DNS subdomain and
fails against MinIO.
The SparkMLPipeline class then shows two canonical tasks. process_text_data() reads raw text with
spark.read.text, applies a lowercase and regexp-based cleanup, filters out near-empty rows, stamps each record
with a processing timestamp, and writes the result back to MinIO as Parquet. Writing Parquet matters here - it is
compressed, columnar, and natively splittable, which keeps subsequent Spark jobs fast even as the corpus grows.
create_training_splits() performs a random 80/10/10 split by adding a uniform random column and filtering on it,
then persists each split under a distinct output path.
Random splitting is acceptable for many tasks, but the rand() approach deserves scrutiny: it assigns each row
independently, so rows from the same entity can leak across train and validation sets. For time-series or per-user
data, split on the entity identifier or on a timestamp boundary instead to prevent leakage. The S3A writes also
benefit from Spark’s adaptive query execution (spark.sql.adaptive.enabled), which coalesces shuffle partitions
and keeps the number of small Parquet files under control.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, regexp_replace, lower
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
def create_spark_session(minio_endpoint, access_key, secret_key):
"""Create Spark session configured for MinIO"""
return SparkSession.builder \
.appName("ML Data Processing Pipeline") \
.config("spark.hadoop.fs.s3a.endpoint", minio_endpoint) \
.config("spark.hadoop.fs.s3a.access.key", access_key) \
.config("spark.hadoop.fs.s3a.secret.key", secret_key) \
.config("spark.hadoop.fs.s3a.path.style.access", "true") \
.config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
.config("spark.sql.adaptive.enabled", "true") \
.config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
.getOrCreate()
class SparkMLPipeline:
def __init__(self, spark_session):
self.spark = spark_session
def process_text_data(self, input_path, output_path):
"""Process text data for NLP training"""
# Read raw text data
df = self.spark.read.text(input_path)
# Clean and preprocess
processed_df = df.select(
# Remove special characters and normalize
regexp_replace(lower(col("value")), r'[^\w\s]', '').alias("clean_text")
).filter(
# Filter out empty or very short texts
col("clean_text").isNotNull() & (length(col("clean_text")) > 10)
)
# Add metadata
processed_df = processed_df.withColumn("processed_timestamp", current_timestamp())
# Write processed data
processed_df.write \
.mode("overwrite") \
.parquet(output_path)
return processed_df.count()
def create_training_splits(self, input_path, output_base_path,
train_ratio=0.8, val_ratio=0.1, test_ratio=0.1):
"""Create train/validation/test splits"""
df = self.spark.read.parquet(input_path)
# Add random column for splitting
df_with_rand = df.withColumn("rand", rand())
# Create splits
train_df = df_with_rand.filter(col("rand") < train_ratio)
val_df = df_with_rand.filter(
(col("rand") >= train_ratio) &
(col("rand") < train_ratio + val_ratio)
)
test_df = df_with_rand.filter(col("rand") >= train_ratio + val_ratio)
# Remove random column and save
for split_name, split_df in [("train", train_df), ("val", val_df), ("test", test_df)]:
split_df.drop("rand").write \
.mode("overwrite") \
.parquet(f"{output_base_path}/{split_name}")
return {
"train_count": train_df.count(),
"val_count": val_df.count(),
"test_count": test_df.count()
}
# Usage example
spark = create_spark_session(
"http://minio:9000", "minioadmin", "minioadmin"
)
pipeline = SparkMLPipeline(spark)
# Process text data
text_count = pipeline.process_text_data(
"s3a://raw-data/text/reviews/",
"s3a://bronze-data/text/processed_reviews/"
)
# Create training splits
split_counts = pipeline.create_training_splits(
"s3a://bronze-data/text/processed_reviews/",
"s3a://gold-data/datasets/reviews"
)
print(f"Processed {text_count} text samples")
print(f"Split counts: {split_counts}")
The Spark integration is the workhorse of the pipeline: it converts messy raw objects into clean, partitioned
training data entirely through S3A reads and writes. For production workloads, add an object-store-aware shuffle
service, enable S3A’s fast upload path (fs.s3a.fast.upload), and size the number of output partitions to match
your cluster’s read parallelism so downstream jobs do not fight over small files.
Several details in the Spark session configuration are easy to gloss over but critical in practice. The
fs.s3a.path.style.access=true flag is non-negotiable for MinIO: S3’s default virtual-hosted addressing assumes
the bucket is a DNS subdomain of the endpoint, which only works on real AWS, whereas MinIO always addresses buckets
as path segments. The explicit fs.s3a.impl value pins the filesystem class so that Spark does not try to
auto-detect a different connector, and the two adaptive execution settings let the query planner merge shuffle
partitions dynamically, which matters because the number of partitions produced by a distributed shuffle is rarely
optimal by default.
The process_text_data() method illustrates a clean read-transform-write cycle that stays entirely within the S3A
namespace. The pipeline expression chains three operations in a single select: lowercasing, removing
non-alphanumeric characters via the regex [^\w\s], and aliasing the result as clean_text. A filter then drops
null or very short rows, current_timestamp() stamps every row with processing time, and the final write uses
mode("overwrite") so re-runs replace stale output rather than appending duplicates. Returning
processed_df.count() after a distributed write forces Spark to trigger the job and gives the caller a concrete
row count for logging - a small trick that turns an otherwise lazy DataFrame into a measurable side effect.
The create_training_splits() method shows the textbook random-split approach, but it also reveals the classic
pitfall that teams hit at scale. Because every row gets an independent uniform random value, the three filters
produce statistically clean 80/10/10 splits for independent samples - which is exactly right for shuffled, IID
data. But for user-level or time-series data, rows from the same user or the same day can end up in different
splits, silently invalidating any evaluation. The fix is to hash an entity identifier and split on that hash, or
split on a time boundary, which the code comments allude to but do not implement. The method also makes a subtle
correctness choice: it drops the rand column before writing, so the rand partitioning key never leaks into the
persisted datasets.
Finally, notice that both methods pass full s3a:// URIs as paths. This is the elegant payoff of the S3A
configuration - once the session is created with the right endpoint and credentials, MinIO buckets appear to Spark
as ordinary filesystems, and no additional glue code is needed. The same URI scheme that works for raw-data works
for any bucket, which is why the earlier bucket layout pays off: Spark jobs reference buckets by name, and the
entire lake becomes addressable from a single, consistent namespace.
Feature Store Implementation
Building a Production Feature Store
A feature store serves as the central repository for ML features, enabling feature reuse across models, maintaining feature consistency between training and inference, and providing feature lineage tracking. MinIO provides the ideal storage backend for feature stores with its high-performance object storage and S3 compatibility.
import boto3
import json
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional
import hashlib
import pickle
class MinIOFeatureStore:
def __init__(self, endpoint_url, access_key, secret_key):
self.s3 = boto3.client(
's3',
endpoint_url=endpoint_url,
aws_access_key_id=access_key,
aws_secret_access_key=secret_key
)
self.bucket = 'feature-store'
self.metadata_bucket = 'feature-metadata'
self._ensure_buckets()
def _ensure_buckets(self):
"""Ensure feature store buckets exist"""
for bucket in [self.bucket, self.metadata_bucket]:
try:
self.s3.create_bucket(Bucket=bucket)
except:
pass
def register_feature_group(self, group_name, schema, description=""):
"""Register a new feature group with schema"""
feature_group = {
'name': group_name,
'schema': schema,
'description': description,
'created_at': datetime.now().isoformat(),
'version': '1.0',
'feature_count': len(schema),
'data_types': {col: str(dtype) for col, dtype in schema.items()}
}
key = f"feature_groups/{group_name}/metadata.json"
self.s3.put_object(
Bucket=self.metadata_bucket,
Key=key,
Body=json.dumps(feature_group, indent=2),
ContentType='application/json'
)
return feature_group
def compute_and_store_features(self, group_name, entity_id,
features_dict, timestamp=None):
"""Compute and store features for an entity"""
if timestamp is None:
timestamp = datetime.now()
# Add metadata to features
feature_record = {
'entity_id': entity_id,
'features': features_dict,
'timestamp': timestamp.isoformat(),
'computed_at': datetime.now().isoformat(),
'feature_group': group_name,
'version': '1.0'
}
# Create time-partitioned key
date_partition = timestamp.strftime('%Y/%m/%d')
hour_partition = timestamp.strftime('%H')
key = f"features/{group_name}/date={date_partition}/hour={hour_partition}/{entity_id}.json"
self.s3.put_object(
Bucket=self.bucket,
Key=key,
Body=json.dumps(feature_record, indent=2),
ContentType='application/json'
)
return key
def batch_compute_features(self, group_name, entities_features, timestamp=None):
"""Batch compute and store features for multiple entities"""
if timestamp is None:
timestamp = datetime.now()
# Convert to DataFrame for efficient processing
df = pd.DataFrame(entities_features)
# Create Parquet file for batch storage
date_partition = timestamp.strftime('%Y/%m/%d')
hour_partition = timestamp.strftime('%H')
batch_id = hashlib.md5(str(timestamp).encode()).hexdigest()[:8]
key = f"features/{group_name}/date={date_partition}/hour={hour_partition}/batch_{batch_id}.parquet"
# Add metadata columns
df['timestamp'] = timestamp.isoformat()
df['computed_at'] = datetime.now().isoformat()
df['feature_group'] = group_name
df['version'] = '1.0'
# Save as Parquet
parquet_buffer = df.to_parquet(index=False)
self.s3.put_object(
Bucket=self.bucket,
Key=key,
Body=parquet_buffer,
ContentType='application/octet-stream'
)
return key
def get_latest_features(self, group_name, entity_id, max_age_hours=24):
"""Retrieve latest features for an entity"""
# Search recent partitions
end_time = datetime.now()
start_time = end_time - timedelta(hours=max_age_hours)
# Generate partition prefixes to search
current_time = start_time
prefixes = []
while current_time <= end_time:
date_partition = current_time.strftime('%Y/%m/%d')
hour_partition = current_time.strftime('%H')
prefix = f"features/{group_name}/date={date_partition}/hour={hour_partition}/"
prefixes.append(prefix)
current_time += timedelta(hours=1)
# Search for entity features
latest_features = None
latest_timestamp = None
for prefix in reversed(prefixes): # Start with most recent
try:
response = self.s3.list_objects_v2(
Bucket=self.bucket,
Prefix=prefix
)
for obj in response.get('Contents', []):
key = obj['Key']
# Check if this is our entity (for JSON files)
if key.endswith(f"{entity_id}.json"):
obj_response = self.s3.get_object(Bucket=self.bucket, Key=key)
feature_record = json.loads(obj_response['Body'].read())
record_timestamp = datetime.fromisoformat(feature_record['timestamp'])
if latest_timestamp is None or record_timestamp > latest_timestamp:
latest_features = feature_record['features']
latest_timestamp = record_timestamp
# Check batch Parquet files
elif key.endswith('.parquet'):
# Download and search Parquet file
local_path = f"/tmp/{os.path.basename(key)}"
self.s3.download_file(self.bucket, key, local_path)
df = pd.read_parquet(local_path)
entity_rows = df[df['entity_id'] == entity_id]
if not entity_rows.empty:
# Get most recent row for this entity
latest_row = entity_rows.loc[entity_rows['timestamp'].idxmax()]
row_timestamp = datetime.fromisoformat(latest_row['timestamp'])
if latest_timestamp is None or row_timestamp > latest_timestamp:
# Extract features (exclude metadata columns)
feature_cols = [col for col in df.columns
if col not in ['entity_id', 'timestamp', 'computed_at',
'feature_group', 'version']]
latest_features = latest_row[feature_cols].to_dict()
latest_timestamp = row_timestamp
os.remove(local_path)
except Exception as e:
print(f"Error searching prefix {prefix}: {e}")
continue
return latest_features, latest_timestamp
def get_historical_features(self, group_name, entity_ids,
start_time, end_time, point_in_time=True):
"""Retrieve historical features for training"""
# Generate time range partitions
current_time = start_time
partitions = []
while current_time <= end_time:
date_partition = current_time.strftime('%Y/%m/%d')
partitions.append(f"features/{group_name}/date={date_partition}/")
current_time += timedelta(days=1)
all_features = []
for partition in partitions:
try:
response = self.s3.list_objects_v2(
Bucket=self.bucket,
Prefix=partition
)
for obj in response.get('Contents', []):
key = obj['Key']
if key.endswith('.parquet'):
# Process Parquet files
local_path = f"/tmp/{os.path.basename(key)}"
self.s3.download_file(self.bucket, key, local_path)
df = pd.read_parquet(local_path)
# Filter by entity IDs and time range
filtered_df = df[
(df['entity_id'].isin(entity_ids)) &
(pd.to_datetime(df['timestamp']) >= start_time) &
(pd.to_datetime(df['timestamp']) <= end_time)
]
if not filtered_df.empty:
all_features.append(filtered_df)
os.remove(local_path)
except Exception as e:
print(f"Error processing partition {partition}: {e}")
continue
if all_features:
combined_df = pd.concat(all_features, ignore_index=True)
if point_in_time:
# For point-in-time correctness, get latest features before each timestamp
combined_df = combined_df.sort_values(['entity_id', 'timestamp'])
return combined_df.groupby('entity_id').last().reset_index()
else:
return combined_df
return pd.DataFrame()
# Example: User behavior features
def compute_user_features(user_id, user_data, transaction_data):
"""Compute user behavior features"""
features = {}
# Basic user features
features['user_age'] = user_data.get('age', 0)
features['user_tenure_days'] = (datetime.now() - user_data['signup_date']).days
features['user_tier'] = user_data.get('tier', 'basic')
# Transaction-based features
if transaction_data:
features['total_transactions'] = len(transaction_data)
features['total_amount'] = sum(t['amount'] for t in transaction_data)
features['avg_transaction_amount'] = features['total_amount'] / features['total_transactions']
features['days_since_last_transaction'] = (
datetime.now() - max(t['timestamp'] for t in transaction_data)
).days
# Category preferences
categories = [t['category'] for t in transaction_data]
category_counts = pd.Series(categories).value_counts()
features['top_category'] = category_counts.index[0] if not category_counts.empty else 'unknown'
features['category_diversity'] = len(category_counts)
else:
features.update({
'total_transactions': 0,
'total_amount': 0.0,
'avg_transaction_amount': 0.0,
'days_since_last_transaction': 999,
'top_category': 'unknown',
'category_diversity': 0
})
return features
# Usage example
feature_store = MinIOFeatureStore(
'http://minio:9000', 'minioadmin', 'minioadmin'
)
# Register feature group
user_schema = {
'user_age': int,
'user_tenure_days': int,
'user_tier': str,
'total_transactions': int,
'total_amount': float,
'avg_transaction_amount': float,
'days_since_last_transaction': int,
'top_category': str,
'category_diversity': int
}
feature_store.register_feature_group(
'user_behavior',
user_schema,
'User behavior and transaction features'
)
# Compute and store features
user_data = {'age': 28, 'signup_date': datetime(2024, 1, 1), 'tier': 'premium'}
transaction_data = [
{'amount': 50.0, 'category': 'food', 'timestamp': datetime(2026, 1, 1)},
{'amount': 200.0, 'category': 'electronics', 'timestamp': datetime(2026, 1, 15)}
]
features = compute_user_features('user_123', user_data, transaction_data)
feature_store.compute_and_store_features('user_behavior', 'user_123', features)
# Retrieve latest features
latest_features, timestamp = feature_store.get_latest_features('user_behavior', 'user_123')
print(f"Latest features for user_123: {latest_features}")
The feature store closes the training-serving skew gap: features computed and persisted once are retrieved
identically at training time (via get_historical_features) and at serving time (via get_latest_features). The
time-partitioned Parquet layout keeps batch scans efficient, and the point-in-time join logic prevents label
leakage by ensuring every training row uses only features available before its target timestamp - the same
discipline that dedicated platforms like Feast enforce.
Looking at the storage model more closely, the feature store makes two deliberate layout choices. First, it
separates feature data from feature metadata into distinct buckets: the feature-store bucket holds the actual
computed feature values, while feature-metadata holds the schema definitions and group registrations. This
separation means the metadata catalog can be read quickly without scanning feature data, and it also lets you apply
different lifecycle policies - keeping metadata indefinitely while expiring old feature partitions. The schema
registered in register_feature_group() records data types for every column, which gives the store a
self-describing catalog that any consumer can introspect before training.
Second, the store supports two storage formats with a single key layout. Single-entity writes go to JSON objects
named {entity_id}.json, ideal for the point lookups that online serving performs, while
batch_compute_features() writes columnar Parquet files with a batch ID in the name, ideal for the full scans that
offline training requires. Both formats sit under the same features/{group}/date=.../hour=... prefix, so a single
listing can discover both. The get_latest_features() retrieval method exploits exactly this hybrid layout,
checking JSON per-entity files first and then falling back to scanning recent Parquet batches for the entity’s most
recent row.
The time-partitioning scheme deserves special attention because it is what makes both retrieval methods practical.
Features are stored under date=YYYY/MM/DD/hour=HH, which means get_latest_features() can bound its search
window to the last 24 hours by enumerating just 24 hourly prefixes instead of scanning the entire store.
get_historical_features() does the opposite - it walks day-by-day partitions over an arbitrary historical range
and merges the results, with a point-in-time flag that keeps only the last feature value per entity before each
training timestamp. This is precisely the semantics that online/offline consistency requires, and it is remarkable
how much of a feature platform is expressible with nothing more than a well-chosen key prefix.
The example compute_user_features() function shows the kind of domain logic that feeds the store: it derives
aggregates from raw transaction records - counts, sums, averages, recency, category preferences - and fills in
sensible defaults (0, 0.0, 999) when no transactions exist. Notice how the function guards against division
by zero by checking total_transactions before computing the average; such defensive defaults are essential in
feature engineering, because a missing or empty input is a normal runtime condition, not an exceptional one. The
result is that the store always receives a complete, typed feature record, and consumers never have to handle
partial dicts.
Feature Engineering Pipeline
Raw features are rarely sufficient for a good model; most teams derive additional features that encode domain
knowledge. The FeatureEngineeringPipeline class below turns this derivation into a small, declarative system:
transformers are registered by name with an optional list of dependencies, and compute_derived_features()
executes them in dependency order, skipping any whose inputs are not yet available. This mirrors the behavior of
tools like Feast and Tecton but keeps the implementation lightweight and entirely on top of MinIO.
The example transformers show two common feature families. compute_ratio_features() creates normalized features
such as amount_per_transaction and transactions_per_day, which remove the effect of scale and are far more
robust inputs for models like logistic regression or neural networks. compute_categorical_features() one-hot
encodes both the user tier and the top spending category, producing binary columns that the model can consume
directly. Declaring that the categorical transformer depends on the ratio transformer ensures a stable execution
order as the feature set grows.
Executing features in dependency order matters for correctness: a feature that depends on another must wait until
its input is computed, and the executed set enforces this without any global scheduler. In production you would
persist the computed features back through the feature store’s compute_and_store_features() call, so that the
same values are available at training time and at inference time - preventing the classic training-serving skew
that occurs when features are recomputed differently in the two paths.
class FeatureEngineeringPipeline:
def __init__(self, feature_store):
self.feature_store = feature_store
self.transformers = {}
def register_transformer(self, name, transform_func, dependencies=None):
"""Register a feature transformation function"""
self.transformers[name] = {
'func': transform_func,
'dependencies': dependencies or []
}
def compute_derived_features(self, base_features):
"""Compute derived features from base features"""
derived_features = base_features.copy()
# Execute transformers in dependency order
executed = set()
while len(executed) < len(self.transformers):
for name, transformer in self.transformers.items():
if name in executed:
continue
# Check if dependencies are satisfied
deps_satisfied = all(dep in executed for dep in transformer['dependencies'])
if deps_satisfied:
try:
new_features = transformer'func'
derived_features.update(new_features)
executed.add(name)
except Exception as e:
print(f"Error executing transformer {name}: {e}")
executed.add(name) # Skip this transformer
return derived_features
# Example transformers
def compute_ratio_features(features):
"""Compute ratio-based features"""
ratios = {}
if features.get('total_transactions', 0) > 0:
ratios['amount_per_transaction'] = (
features['total_amount'] / features['total_transactions']
)
if features.get('user_tenure_days', 0) > 0:
ratios['transactions_per_day'] = (
features['total_transactions'] / features['user_tenure_days']
)
return ratios
def compute_categorical_features(features):
"""Compute categorical encodings"""
categorical = {}
# One-hot encode user tier
tiers = ['basic', 'premium', 'enterprise']
for tier in tiers:
categorical[f'tier_{tier}'] = 1 if features.get('user_tier') == tier else 0
# Encode top category
categories = ['food', 'electronics', 'clothing', 'books', 'other']
for category in categories:
categorical[f'top_category_{category}'] = (
1 if features.get('top_category') == category else 0
)
return categorical
# Setup pipeline
pipeline = FeatureEngineeringPipeline(feature_store)
pipeline.register_transformer('ratios', compute_ratio_features)
pipeline.register_transformer('categorical', compute_categorical_features, ['ratios'])
# Compute enhanced features
base_features = {'user_age': 28, 'total_transactions': 10, 'total_amount': 500.0}
enhanced_features = pipeline.compute_derived_features(base_features)
print(f"Enhanced features: {enhanced_features}")
A dependency-ordered transformer registry is a miniature feature-engineering platform. It keeps feature logic discoverable, reusable, and testable, and it composes cleanly with the feature store: register the group, compute derived features, persist them, and both training and serving read identical values. Log the transformation versions into the lineage tracker so that any model can be reproduced from the exact feature code that produced its inputs.
The execution algorithm inside compute_derived_features() is worth tracing, because it is the machinery that
makes the dependency system work. Each transformer is registered with a name and an optional list of dependency
names, and the method loops until every transformer has executed. In each pass it scans the registry for a
transformer whose dependencies are all in the executed set, runs it, merges its outputs into the feature dict,
and records it as done. This is effectively a topological sort on the dependency graph, expressed in a few lines of
plain Python with no graph library. If two transformers have no dependencies on each other, either may run first -
order only matters where dependencies declare it, which is exactly the correct semantics.
There are two practical weaknesses in the reference implementation that production code should address. The first
is the accidental transformer'func' typo in the call site, which would raise a NameError at runtime; the
surrounding try/except swallows it and marks the transformer as executed, meaning the derived features silently
miss that transformer’s contribution. This is a cautionary example of how a broad except can mask real bugs - a
production version should log the traceback and surface the failure rather than skip it. The second weakness is
that a circular dependency would cause an infinite loop, since no transformer would ever become eligible; adding a
cycle detector or an iteration cap would turn that silent hang into a clear error.
The two example transformers also illustrate a trade-off between feature families. Ratio features like
amount_per_transaction are continuous, scale-free, and information-dense, making them excellent inputs for linear
models, tree ensembles, and neural networks alike. One-hot features like tier_premium and top_category_food are
discrete and sparse, and they multiply the feature dimension rapidly - with dozens of categories, the vector
becomes mostly zeros, which can slow training and dilute signal. For high-cardinality categoricals, modern practice
leans toward embeddings, feature hashing, or target encoding instead of strict one-hot encoding, but the one-hot
form shown here remains the simplest correct baseline and is easy to swap later.
Finally, notice that this pipeline is stateless: compute_derived_features() takes base features and returns
derived features without touching MinIO. That purity is intentional, because it keeps the transformation logic
unit-testable in isolation. Persistence is a separate concern handled by the feature store’s write methods, so the
two systems compose cleanly - transformers compute, the store persists, and the registry’s lineage records tie the
whole chain together. This separation is a good template for feature engineering in general: keep pure functions
for computation and let dedicated storage and metadata layers handle durability.
Model Storage and Registry
Production Model Registry
A robust model registry manages the complete lifecycle of ML models, from training artifacts to production deployments. MinIO provides the storage foundation for model artifacts, metadata, and experiment tracking.
The MLModelRegistry class below treats every model as a directory under models/{name}/{version}/ containing a
metadata.json file plus any number of artifacts - PyTorch state dicts, pickled preprocessors, and joblib
estimators. The register_model() method serializes in-memory objects using the format implied by their file
extension (.pt via torch.save, .pkl via pickle, .joblib via joblib.dump) and uploads the bytes, while
also accepting already-serialized local files. Writing the rich metadata document - framework, metrics, parameters,
training data, dependencies, tags - alongside the artifacts is what turns MinIO from a folder of weights into a
true registry, because every downstream system can resolve a model without any side-channel knowledge.
The registry also implements the operations a production workflow needs: list_models() and
list_model_versions() reconstruct the model tree from S3 common prefixes, get_latest_version() applies semantic
version ordering, and promote_model() moves a version through lifecycle stages such as staging and production by
mutating its metadata. load_model() reverses registration, reconstructing each artifact according to its
extension and mapping everything to CPU memory so that any serving machine can instantiate the model. Crucially,
register_model() builds a unique model_id from name, version, and timestamp, giving every deployment an
unambiguous handle for tracing.
Using the object store as the registry has a clear trade-off versus platforms like MLflow or Weights & Biases: you
manage the versioning semantics yourself rather than inheriting them. In exchange, you gain a registry that is
trivially portable, has no extra servers to operate, and is readable by every S3 tool. The second class in this
section, ModelCheckpointManager, demonstrates the same pattern for the frequent checkpoints produced during
training, storing model and optimizer state along with per-checkpoint metadata so that crashed jobs can resume from
any epoch.
import boto3
import json
import torch
import pickle
import joblib
from datetime import datetime
from typing import Dict, Any, Optional, List
import hashlib
import os
class MLModelRegistry:
def __init__(self, endpoint_url, access_key, secret_key):
self.s3 = boto3.client(
's3',
endpoint_url=endpoint_url,
aws_access_key_id=access_key,
aws_secret_access_key=secret_key
)
self.models_bucket = 'model-registry'
self.experiments_bucket = 'experiment-tracking'
self._ensure_buckets()
def _ensure_buckets(self):
"""Ensure model registry buckets exist"""
for bucket in [self.models_bucket, self.experiments_bucket]:
try:
self.s3.create_bucket(Bucket=bucket)
except:
pass
def register_model(self, model_name: str, version: str,
model_artifacts: Dict[str, Any],
metadata: Dict[str, Any]) -> str:
"""Register a new model version with artifacts and metadata"""
model_id = f"{model_name}_{version}_{int(datetime.now().timestamp())}"
model_path = f"models/{model_name}/{version}"
# Store model artifacts
artifact_paths = {}
for artifact_name, artifact_data in model_artifacts.items():
artifact_key = f"{model_path}/artifacts/{artifact_name}"
if isinstance(artifact_data, str) and os.path.exists(artifact_data):
# Upload file
self.s3.upload_file(artifact_data, self.models_bucket, artifact_key)
else:
# Serialize and upload object
if artifact_name.endswith('.pt') or artifact_name.endswith('.pth'):
# PyTorch model
with open(f'/tmp/{artifact_name}', 'wb') as f:
torch.save(artifact_data, f)
self.s3.upload_file(f'/tmp/{artifact_name}', self.models_bucket, artifact_key)
os.remove(f'/tmp/{artifact_name}')
elif artifact_name.endswith('.pkl'):
# Pickle object
with open(f'/tmp/{artifact_name}', 'wb') as f:
pickle.dump(artifact_data, f)
self.s3.upload_file(f'/tmp/{artifact_name}', self.models_bucket, artifact_key)
os.remove(f'/tmp/{artifact_name}')
elif artifact_name.endswith('.joblib'):
# Joblib object
joblib.dump(artifact_data, f'/tmp/{artifact_name}')
self.s3.upload_file(f'/tmp/{artifact_name}', self.models_bucket, artifact_key)
os.remove(f'/tmp/{artifact_name}')
artifact_paths[artifact_name] = f"s3://{self.models_bucket}/{artifact_key}"
# Create model metadata
model_metadata = {
'model_id': model_id,
'name': model_name,
'version': version,
'created_at': datetime.now().isoformat(),
'artifacts': artifact_paths,
'metadata': metadata,
'status': 'registered',
'tags': metadata.get('tags', []),
'framework': metadata.get('framework', 'unknown'),
'model_type': metadata.get('model_type', 'unknown'),
'metrics': metadata.get('metrics', {}),
'parameters': metadata.get('parameters', {}),
'training_data': metadata.get('training_data', {}),
'dependencies': metadata.get('dependencies', [])
}
# Store metadata
metadata_key = f"{model_path}/metadata.json"
self.s3.put_object(
Bucket=self.models_bucket,
Key=metadata_key,
Body=json.dumps(model_metadata, indent=2),
ContentType='application/json'
)
return model_id
def load_model(self, model_name: str, version: str = 'latest') -> Dict[str, Any]:
"""Load model artifacts and metadata"""
if version == 'latest':
version = self.get_latest_version(model_name)
model_path = f"models/{model_name}/{version}"
# Load metadata
metadata_key = f"{model_path}/metadata.json"
try:
metadata_obj = self.s3.get_object(Bucket=self.models_bucket, Key=metadata_key)
metadata = json.loads(metadata_obj['Body'].read())
except Exception as e:
raise ValueError(f"Model {model_name} version {version} not found: {e}")
# Load artifacts
artifacts = {}
for artifact_name, artifact_path in metadata['artifacts'].items():
artifact_key = artifact_path.replace(f"s3://{self.models_bucket}/", "")
local_path = f"/tmp/{artifact_name}"
self.s3.download_file(self.models_bucket, artifact_key, local_path)
# Load based on file extension
if artifact_name.endswith('.pt') or artifact_name.endswith('.pth'):
artifacts[artifact_name] = torch.load(local_path, map_location='cpu')
elif artifact_name.endswith('.pkl'):
with open(local_path, 'rb') as f:
artifacts[artifact_name] = pickle.load(f)
elif artifact_name.endswith('.joblib'):
artifacts[artifact_name] = joblib.load(local_path)
else:
# Keep as file path for other formats
artifacts[artifact_name] = local_path
return {
'metadata': metadata,
'artifacts': artifacts
}
def list_models(self, name_filter: Optional[str] = None) -> List[Dict[str, Any]]:
"""List all registered models"""
models = []
try:
response = self.s3.list_objects_v2(
Bucket=self.models_bucket,
Prefix='models/',
Delimiter='/'
)
# Get model names from common prefixes
for prefix in response.get('CommonPrefixes', []):
model_name = prefix['Prefix'].split('/')[-2]
if name_filter and name_filter not in model_name:
continue
# Get versions for this model
versions = self.list_model_versions(model_name)
models.append({
'name': model_name,
'versions': versions,
'latest_version': max(versions) if versions else None
})
except Exception as e:
print(f"Error listing models: {e}")
return models
def list_model_versions(self, model_name: str) -> List[str]:
"""List all versions of a specific model"""
versions = []
try:
response = self.s3.list_objects_v2(
Bucket=self.models_bucket,
Prefix=f'models/{model_name}/',
Delimiter='/'
)
for prefix in response.get('CommonPrefixes', []):
version = prefix['Prefix'].split('/')[-2]
versions.append(version)
except Exception as e:
print(f"Error listing versions for {model_name}: {e}")
return sorted(versions)
def get_latest_version(self, model_name: str) -> str:
"""Get the latest version of a model"""
versions = self.list_model_versions(model_name)
if not versions:
raise ValueError(f"No versions found for model {model_name}")
# Sort versions (assuming semantic versioning)
try:
sorted_versions = sorted(versions, key=lambda v: [int(x) for x in v.split('.')])
return sorted_versions[-1]
except:
# Fallback to string sorting
return sorted(versions)[-1]
def promote_model(self, model_name: str, version: str, stage: str) -> bool:
"""Promote model to a specific stage (staging, production, etc.)"""
model_path = f"models/{model_name}/{version}"
metadata_key = f"{model_path}/metadata.json"
try:
# Load current metadata
metadata_obj = self.s3.get_object(Bucket=self.models_bucket, Key=metadata_key)
metadata = json.loads(metadata_obj['Body'].read())
# Update stage
metadata['stage'] = stage
metadata['promoted_at'] = datetime.now().isoformat()
# Save updated metadata
self.s3.put_object(
Bucket=self.models_bucket,
Key=metadata_key,
Body=json.dumps(metadata, indent=2),
ContentType='application/json'
)
return True
except Exception as e:
print(f"Error promoting model: {e}")
return False
# Advanced model checkpoint management
class ModelCheckpointManager:
def __init__(self, s3_client, bucket='model-checkpoints'):
self.s3 = s3_client
self.bucket = bucket
def save_checkpoint(self, model, optimizer, epoch, metrics,
experiment_id, checkpoint_name=None):
"""Save training checkpoint with comprehensive metadata"""
if checkpoint_name is None:
checkpoint_name = f"checkpoint_epoch_{epoch}"
# Create checkpoint data
checkpoint_data = {
'epoch': epoch,
'model_state_dict': model.state_dict(),
'optimizer_state_dict': optimizer.state_dict(),
'metrics': metrics,
'timestamp': datetime.now().isoformat(),
'experiment_id': experiment_id
}
# Save checkpoint
checkpoint_path = f"/tmp/{checkpoint_name}.pt"
torch.save(checkpoint_data, checkpoint_path)
# Upload to MinIO
key = f"experiments/{experiment_id}/checkpoints/{checkpoint_name}.pt"
self.s3.upload_file(checkpoint_path, self.bucket, key)
# Save checkpoint metadata
metadata = {
'checkpoint_name': checkpoint_name,
'experiment_id': experiment_id,
'epoch': epoch,
'metrics': metrics,
'timestamp': datetime.now().isoformat(),
'model_architecture': str(model),
'optimizer_config': str(optimizer),
's3_path': f"s3://{self.bucket}/{key}"
}
metadata_key = f"experiments/{experiment_id}/checkpoints/{checkpoint_name}_metadata.json"
self.s3.put_object(
Bucket=self.bucket,
Key=metadata_key,
Body=json.dumps(metadata, indent=2),
ContentType='application/json'
)
# Cleanup local file
os.remove(checkpoint_path)
return key
def load_checkpoint(self, experiment_id, checkpoint_name):
"""Load checkpoint from MinIO"""
key = f"experiments/{experiment_id}/checkpoints/{checkpoint_name}.pt"
local_path = f"/tmp/{checkpoint_name}.pt"
self.s3.download_file(self.bucket, key, local_path)
checkpoint = torch.load(local_path, map_location='cpu')
os.remove(local_path)
return checkpoint
def list_checkpoints(self, experiment_id):
"""List all checkpoints for an experiment"""
prefix = f"experiments/{experiment_id}/checkpoints/"
response = self.s3.list_objects_v2(
Bucket=self.bucket,
Prefix=prefix
)
checkpoints = []
for obj in response.get('Contents', []):
if obj['Key'].endswith('.pt'):
checkpoint_name = os.path.basename(obj['Key']).replace('.pt', '')
checkpoints.append({
'name': checkpoint_name,
'key': obj['Key'],
'size': obj['Size'],
'last_modified': obj['LastModified']
})
return sorted(checkpoints, key=lambda x: x['last_modified'], reverse=True)
# Usage examples
registry = MLModelRegistry('http://minio:9000', 'minioadmin', 'minioadmin')
# Register a PyTorch model
model = torch.nn.Linear(10, 1)
optimizer = torch.optim.Adam(model.parameters())
model_artifacts = {
'model.pt': model.state_dict(),
'optimizer.pt': optimizer.state_dict(),
'preprocessing.pkl': {'scaler': 'StandardScaler', 'params': {}}
}
metadata = {
'framework': 'pytorch',
'model_type': 'linear_regression',
'metrics': {'mse': 0.05, 'r2': 0.95},
'parameters': {'learning_rate': 0.001, 'batch_size': 32},
'training_data': {'dataset': 'housing_prices', 'samples': 10000},
'tags': ['regression', 'production-ready']
}
model_id = registry.register_model('housing_price_predictor', 'v1.0', model_artifacts, metadata)
print(f"Registered model: {model_id}")
# Load model
loaded_model = registry.load_model('housing_price_predictor', 'v1.0')
print(f"Loaded model metadata: {loaded_model['metadata']['metrics']}")
# List all models
models = registry.list_models()
for model in models:
print(f"Model: {model['name']}, Versions: {model['versions']}")
Together, the registry and checkpoint manager cover the full persistence spectrum of model management: coarse
versioned deployments in the model-registry bucket and fine-grained resume points in the model-checkpoints
bucket. Hook the checkpoint manager into your training loop’s scheduler to save after every epoch (or on
improvement), and promote to the registry only when validation metrics qualify. Because both layers are plain S3
objects, a production serving stack can load any version or resume any run with a single bucket prefix - no
migration, no lock-in.
The register_model() method contains the core insight of the whole registry: a model is not a single file but a
bundle of artifacts plus a metadata envelope. The artifacts dict accepts either local file paths (detected via
os.path.exists) or in-memory Python objects, and the method serializes each according to its file extension. That
dispatch - torch.save for .pt, pickle for .pkl, joblib.dump for .joblib - is what lets one registry
handle models trained in completely different frameworks without any framework-specific branching in the calling
code. The resulting artifacts map records each file’s s3:// URI, so the metadata document becomes a complete
manifest of where every piece of the model lives.
The metadata envelope itself is more than documentation; it is the contract that every downstream consumer relies
on. list_models() walks the models/ prefix using Delimiter='/' to discover model names and versions purely
from directory structure, and get_latest_version() sorts version strings using semantic-versioning-aware
comparison, falling back to string sorting if a version is not semver-shaped. promote_model() reads the metadata,
updates the stage field to values like staging or production, and writes it back - an approach that gives you
MLflow-style stage transitions without any centralized service. The load_model() counterpart downloads each
artifact and reconstructs it with map_location='cpu', so models can be loaded on any machine regardless of where
they were trained.
The ModelCheckpointManager then shows the complementary concern: saving state frequently and cheaply during
training. Instead of bundling a full model, it captures the exact snapshot needed to resume - model.state_dict(),
optimizer.state_dict(), the epoch number, metrics, and experiment ID - and saves it to a checkpoint file plus a
human-readable JSON metadata record. The s3_path field embedded in that metadata means the checkpoint is fully
self-describing: anyone who finds the JSON can find and reconstruct the corresponding .pt snapshot. This is the
pattern that makes interrupted long-running training jobs recoverable without losing hours of GPU time.
One important production note: the registry code saves temporary files under /tmp and removes them after upload,
which works well but is not crash-safe. If the process dies between torch.save and the S3 upload, a temp file
leaks and the checkpoint is lost. Production implementations typically stream serialized bytes directly to
put_object where the format allows it, or use a shared ephemeral volume with cleanup. The fundamental
architecture - versioned artifact bundles plus metadata manifests, all in object storage - is sound and widely
used; the durability details around temporary files are the part to harden.
Integration with PyTorch and TensorFlow
PyTorch Integration with MinIO
PyTorch’s DataLoader can be seamlessly integrated with MinIO for efficient training data loading. This enables distributed training scenarios where data is stored centrally in MinIO and accessed by multiple training nodes.
The MinIOImageDataset below demonstrates the key pattern: a custom torch.utils.data.Dataset that fetches bytes
from object storage on demand instead of assuming a local filesystem. It resolves the sample list either from an
explicit JSON manifest or by paginating the bucket and deriving labels from directory structure, then serves images
through a small in-memory cache that holds the most recent cache_size objects. Caching is critical - naive
per-sample GETs over object storage would stall the GPU, so the cache absorbs the repeat reads that shuffling
produces, and the blank-image fallback keeps a corrupted object from killing the whole epoch.
MinIOStreamingDataset addresses the other end of the spectrum: datasets too large to index or cache. As an
IterableDataset, it streams one object at a time, splits the key list across DataLoader workers for distributed
training, and shuffles each worker’s shard - a coarse but effective substitute for global shuffling. Between these
two classes you can serve everything from a few thousand images to a multi-terabyte corpus from the same bucket
layout.
PyTorchMinIOTrainer closes the loop by wiring the datasets into real training: it applies ImageNet-style
augmentations, moves batches to CUDA, runs standard forward/backward passes, and reports per-epoch train and
validation accuracy. The train and validation prefixes (train/, val/) map directly onto the split directories
produced earlier by the Spark pipeline, so the data lake and the training loop are the same system. The obvious
caveat is that per-sample network I/O is slower than local reads, which is why high-throughput production training
typically pairs a streaming dataset with a local prefetch cache such as FSSpec-backed WebDataset.
import torch
from torch.utils.data import Dataset, DataLoader, IterableDataset
import boto3
from PIL import Image
import io
import json
import pandas as pd
from torchvision import transforms
import numpy as np
class MinIOImageDataset(Dataset):
def __init__(self, bucket, prefix, s3_client, transform=None,
cache_size=1000, metadata_file=None):
self.s3 = s3_client
self.bucket = bucket
self.prefix = prefix
self.transform = transform
self.cache = {}
self.cache_size = cache_size
# Load image paths and labels
self.samples = self._load_samples(metadata_file)
def _load_samples(self, metadata_file):
"""Load sample paths and labels from metadata or S3 listing"""
samples = []
if metadata_file:
# Load from metadata file
obj = self.s3.get_object(Bucket=self.bucket, Key=metadata_file)
metadata = json.loads(obj['Body'].read())
for item in metadata['samples']:
samples.append({
'key': item['path'],
'label': item['label'],
'metadata': item.get('metadata', {})
})
else:
# List objects from S3
paginator = self.s3.get_paginator('list_objects_v2')
for page in paginator.paginate(Bucket=self.bucket, Prefix=self.prefix):
for obj in page.get('Contents', []):
key = obj['Key']
if key.lower().endswith(('.jpg', '.jpeg', '.png')):
# Extract label from path (assuming path structure like class/image.jpg)
path_parts = key.split('/')
label = path_parts[-2] if len(path_parts) > 1 else 'unknown'
samples.append({
'key': key,
'label': label,
'metadata': {}
})
return samples
def __len__(self):
return len(self.samples)
def __getitem__(self, idx):
sample = self.samples[idx]
key = sample['key']
# Check cache first
if key in self.cache:
image_data = self.cache[key]
else:
# Download from MinIO
try:
obj = self.s3.get_object(Bucket=self.bucket, Key=key)
image_data = obj['Body'].read()
# Cache if under limit
if len(self.cache) < self.cache_size:
self.cache[key] = image_data
except Exception as e:
print(f"Error loading {key}: {e}")
# Return a blank image as fallback
image_data = Image.new('RGB', (224, 224)).tobytes()
# Convert to PIL Image
try:
image = Image.open(io.BytesIO(image_data)).convert('RGB')
except Exception as e:
print(f"Error processing image {key}: {e}")
image = Image.new('RGB', (224, 224))
# Apply transforms
if self.transform:
image = self.transform(image)
# Convert label to tensor
if isinstance(sample['label'], str):
# For classification, you'd have a label mapping
label_map = getattr(self, 'label_map', {})
label = label_map.get(sample['label'], 0)
else:
label = sample['label']
return image, torch.tensor(label, dtype=torch.long)
class MinIOStreamingDataset(IterableDataset):
"""Streaming dataset for very large datasets that don't fit in memory"""
def __init__(self, bucket, prefix, s3_client, transform=None,
batch_size=32, shuffle_buffer_size=10000):
self.s3 = s3_client
self.bucket = bucket
self.prefix = prefix
self.transform = transform
self.batch_size = batch_size
self.shuffle_buffer_size = shuffle_buffer_size
def __iter__(self):
# Get worker info for distributed training
worker_info = torch.utils.data.get_worker_info()
# List all objects
paginator = self.s3.get_paginator('list_objects_v2')
all_keys = []
for page in paginator.paginate(Bucket=self.bucket, Prefix=self.prefix):
for obj in page.get('Contents', []):
key = obj['Key']
if key.lower().endswith(('.jpg', '.jpeg', '.png')):
all_keys.append(key)
# Distribute keys across workers
if worker_info is not None:
per_worker = len(all_keys) // worker_info.num_workers
start_idx = worker_info.id * per_worker
end_idx = start_idx + per_worker if worker_info.id < worker_info.num_workers - 1 else len(all_keys)
keys = all_keys[start_idx:end_idx]
else:
keys = all_keys
# Shuffle keys
np.random.shuffle(keys)
# Stream data
for key in keys:
try:
obj = self.s3.get_object(Bucket=self.bucket, Key=key)
image_data = obj['Body'].read()
image = Image.open(io.BytesIO(image_data)).convert('RGB')
if self.transform:
image = self.transform(image)
# Extract label from path
label = key.split('/')[-2]
yield image, label
except Exception as e:
print(f"Error streaming {key}: {e}")
continue
# PyTorch training loop with MinIO
class PyTorchMinIOTrainer:
def __init__(self, model, s3_client, bucket):
self.model = model
self.s3 = s3_client
self.bucket = bucket
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
self.model.to(self.device)
def create_data_loaders(self, train_prefix, val_prefix, batch_size=32, num_workers=4):
"""Create training and validation data loaders"""
# Define transforms
train_transform = transforms.Compose([
transforms.Resize((256, 256)),
transforms.RandomCrop(224),
transforms.RandomHorizontalFlip(),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225])
])
val_transform = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225])
])
# Create datasets
train_dataset = MinIOImageDataset(
self.bucket, train_prefix, self.s3, transform=train_transform
)
val_dataset = MinIOImageDataset(
self.bucket, val_prefix, self.s3, transform=val_transform
)
# Create data loaders
train_loader = DataLoader(
train_dataset, batch_size=batch_size, shuffle=True,
num_workers=num_workers, pin_memory=True
)
val_loader = DataLoader(
val_dataset, batch_size=batch_size, shuffle=False,
num_workers=num_workers, pin_memory=True
)
return train_loader, val_loader
def train_epoch(self, train_loader, optimizer, criterion):
"""Train for one epoch"""
self.model.train()
total_loss = 0
correct = 0
total = 0
for batch_idx, (data, target) in enumerate(train_loader):
data, target = data.to(self.device), target.to(self.device)
optimizer.zero_grad()
output = self.model(data)
loss = criterion(output, target)
loss.backward()
optimizer.step()
total_loss += loss.item()
pred = output.argmax(dim=1, keepdim=True)
correct += pred.eq(target.view_as(pred)).sum().item()
total += target.size(0)
if batch_idx % 100 == 0:
print(f'Batch {batch_idx}, Loss: {loss.item():.6f}')
return total_loss / len(train_loader), correct / total
def validate(self, val_loader, criterion):
"""Validate model"""
self.model.eval()
val_loss = 0
correct = 0
total = 0
with torch.no_grad():
for data, target in val_loader:
data, target = data.to(self.device), target.to(self.device)
output = self.model(data)
val_loss += criterion(output, target).item()
pred = output.argmax(dim=1, keepdim=True)
correct += pred.eq(target.view_as(pred)).sum().item()
total += target.size(0)
return val_loss / len(val_loader), correct / total
# Usage example
s3_client = boto3.client('s3', endpoint_url='http://minio:9000',
aws_access_key_id='minioadmin',
aws_secret_access_key='minioadmin')
model = torch.hub.load('pytorch/vision:v0.10.0', 'resnet18', pretrained=True)
model.fc = torch.nn.Linear(model.fc.in_features, 10) # 10 classes
trainer = PyTorchMinIOTrainer(model, s3_client, 'training-data')
train_loader, val_loader = trainer.create_data_loaders('train/', 'val/')
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
criterion = torch.nn.CrossEntropyLoss()
# Training loop
for epoch in range(10):
train_loss, train_acc = trainer.train_epoch(train_loader, optimizer, criterion)
val_loss, val_acc = trainer.validate(val_loader, criterion)
print(f'Epoch {epoch}: Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f}, '
f'Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.4f}')
The PyTorch dataset classes map the data lake directly onto the training loop: the same train/ and val/
prefixes produced by the Spark splits are consumed without copying. For single-node experiments the in-memory cache
is enough; for multi-GPU or multi-node training, prefer the streaming dataset with worker-sharded keys so that
MinIO’s parallel I/O is spread across processes, and consider a local prefetch tier to hide object-store latency
behind compute.
Looking at MinIOImageDataset more closely, the design makes a sharp distinction between the index and the data.
The _load_samples() method builds a lightweight in-memory list of {key, label, metadata} tuples - the index -
either from a JSON manifest or by paginating the bucket prefix. Only when __getitem__ is called does the dataset
touch MinIO, downloading a single object’s bytes, converting them to a PIL image, applying the transform, and
returning a tensor. This index-then-fetch split is exactly how production data pipelines are structured: the index
can be cached, shared, or even stored back in MinIO, while each worker fetches exactly the bytes it needs. It also
means len(dataset) is known up front, which DataLoader requires for shuffle=True and progress reporting.
The in-memory cache dict is the most consequential detail for training performance. Object storage round-trips
cost on the order of milliseconds each, which is acceptable for a few images but deadly when multiplied across
thousands of samples per epoch. By caching the raw bytes of the most recently seen cache_size objects, the
dataset turns repeated reads - which shuffle induces - into dict lookups. The cache is bounded, so memory stays
flat, and it stores bytes rather than decoded tensors, which keeps it small and lets the transform pipeline stay
deterministic. The blank-image fallbacks in both the download and decode paths are defensive: a single corrupt
object should degrade one sample, not abort the epoch.
MinIOStreamingDataset solves the opposite problem with a fundamentally different strategy. Because it subclasses
IterableDataset, it does not support random access or true shuffling; instead, its __iter__ yields samples
lazily in the order keys are listed. The worker-aware key sharding - using torch.utils.data.get_worker_info() to
give each DataLoader worker a disjoint slice of the key list - is what makes this scale to multiple processes:
without it, every worker would redundantly list and download the entire dataset. The np.random.shuffle(keys) call
provides a rough shuffle of each worker’s shard, which is statistically adequate for stochastic gradient descent
even though it is not a global shuffle. This streaming class is the right tool when the corpus is larger than any
reasonable cache, and it imposes a much smaller memory footprint.
The PyTorchMinIOTrainer ties it together with a conventional training loop that needs no explanation of the math
but plenty of attention to its I/O behavior. Both loaders are created with num_workers=4 and pin_memory=True;
the former overlaps data loading with GPU compute, while the latter copies tensors into CUDA-friendly pinned memory
to avoid a second copy on transfer. The example swaps the final layer of a pretrained ResNet-18 to a 10-class head,
demonstrating the transfer-learning pattern that dominates real-world computer vision work. Because all of this
reads from the same bucket prefixes the data lake produces, moving from research to production is a matter of
changing prefixes, not rewriting the data pipeline.
TensorFlow Integration with MinIO
TensorFlow’s tf.data API provides excellent integration with cloud storage systems like MinIO through its S3-compatible interface.
The TensorFlowMinIODataset class below wraps MinIO access inside a tf.data.Dataset so the framework’s pipeline
engine handles parallelism, batching, and prefetching. Image loading is expressed with tf.py_function, which
bridges from the graph into a Python closure that downloads bytes from S3, decodes them with
tf.image.decode_image, resizes, and normalizes to the [0, 1] range. Labels are extracted from the path
structure, matching the layout the data lake established earlier. Marking the mapping with
num_parallel_calls=tf.data.AUTOTUNE and adding prefetch(tf.data.AUTOTUNE) lets TensorFlow overlap network I/O
with GPU compute, which is essential for keeping accelerators busy.
The text dataset takes a different route because tokenization is stateful: it builds the dataset from a Python generator and then applies a tokenizer mapping over it. This is a deliberately simplified illustration - in production you would use a proper sub-word tokenizer such as SentencePiece or a Hugging Face tokenizer rather than a raw split, and you would serialize the vocabulary to MinIO so the same mapping is available at serving time.
TensorFlowMinIOTrainer.train_model() pulls it together with standard Keras training: ModelCheckpoint,
EarlyStopping, and ReduceLROnPlateau callbacks protect against overfitting and wasted epochs, and the best
model is saved to disk and uploaded back to MinIO. save_model_artifacts() goes further and packages the model in
the portable SavedModel format, tars it, and uploads it with a JSON metadata record - the TensorFlow counterpart of
the model registry’s PyTorch artifacts. The result is a complete train-and-persist loop in which MinIO is
simultaneously the input source and the output sink.
import tensorflow as tf
import boto3
import json
import numpy as np
from typing import Generator, Tuple
class TensorFlowMinIODataset:
def __init__(self, s3_client, bucket):
self.s3 = s3_client
self.bucket = bucket
def create_image_dataset(self, prefix, batch_size=32, image_size=(224, 224)):
"""Create TensorFlow dataset for images stored in MinIO"""
# Get list of image files
def get_image_paths():
paginator = self.s3.get_paginator('list_objects_v2')
paths = []
for page in paginator.paginate(Bucket=self.bucket, Prefix=prefix):
for obj in page.get('Contents', []):
key = obj['Key']
if key.lower().endswith(('.jpg', '.jpeg', '.png')):
paths.append(key)
return paths
# Create dataset from paths
image_paths = get_image_paths()
def load_and_preprocess_image(path):
"""Load image from MinIO and preprocess"""
# Download image
obj = self.s3.get_object(Bucket=self.bucket, Key=path.numpy().decode())
image_data = obj['Body'].read()
# Decode image
image = tf.image.decode_image(image_data, channels=3)
image = tf.image.resize(image, image_size)
image = tf.cast(image, tf.float32) / 255.0
# Extract label from path
label = path.numpy().decode().split('/')[-2]
return image, label
# Create TensorFlow dataset
dataset = tf.data.Dataset.from_tensor_slices(image_paths)
dataset = dataset.map(
lambda path: tf.py_function(
load_and_preprocess_image,
[path],
[tf.float32, tf.string]
),
num_parallel_calls=tf.data.AUTOTUNE
)
# Batch and prefetch
dataset = dataset.batch(batch_size)
dataset = dataset.prefetch(tf.data.AUTOTUNE)
return dataset
def create_text_dataset(self, prefix, batch_size=32, max_length=512):
"""Create TensorFlow dataset for text data"""
def text_generator():
"""Generator function for text data"""
paginator = self.s3.get_paginator('list_objects_v2')
for page in paginator.paginate(Bucket=self.bucket, Prefix=prefix):
for obj in page.get('Contents', []):
key = obj['Key']
if key.endswith('.txt') or key.endswith('.json'):
try:
obj_data = self.s3.get_object(Bucket=self.bucket, Key=key)
content = obj_data['Body'].read().decode('utf-8')
if key.endswith('.json'):
data = json.loads(content)
text = data.get('text', '')
label = data.get('label', 0)
else:
text = content
label = 0 # Default label
yield text, label
except Exception as e:
print(f"Error processing {key}: {e}")
continue
# Create dataset from generator
dataset = tf.data.Dataset.from_generator(
text_generator,
output_signature=(
tf.TensorSpec(shape=(), dtype=tf.string),
tf.TensorSpec(shape=(), dtype=tf.int32)
)
)
# Tokenize and pad sequences
tokenizer = tf.keras.preprocessing.text.Tokenizer(num_words=10000)
def tokenize_text(text, label):
# This is a simplified tokenization - in practice, use a proper tokenizer
tokens = tf.strings.split(text)
# Convert to indices (simplified)
return tokens, label
dataset = dataset.map(tokenize_text)
dataset = dataset.batch(batch_size)
dataset = dataset.prefetch(tf.data.AUTOTUNE)
return dataset
class TensorFlowMinIOTrainer:
def __init__(self, model, s3_client, bucket):
self.model = model
self.s3 = s3_client
self.bucket = bucket
self.dataset_creator = TensorFlowMinIODataset(s3_client, bucket)
def train_model(self, train_prefix, val_prefix, epochs=10, batch_size=32):
"""Train model with data from MinIO"""
# Create datasets
train_dataset = self.dataset_creator.create_image_dataset(
train_prefix, batch_size
)
val_dataset = self.dataset_creator.create_image_dataset(
val_prefix, batch_size
)
# Compile model
self.model.compile(
optimizer='adam',
loss='sparse_categorical_crossentropy',
metrics=['accuracy']
)
# Setup callbacks
callbacks = [
tf.keras.callbacks.ModelCheckpoint(
'/tmp/best_model.h5',
save_best_only=True,
monitor='val_accuracy'
),
tf.keras.callbacks.EarlyStopping(
patience=3,
monitor='val_loss'
),
tf.keras.callbacks.ReduceLROnPlateau(
factor=0.5,
patience=2,
monitor='val_loss'
)
]
# Train model
history = self.model.fit(
train_dataset,
validation_data=val_dataset,
epochs=epochs,
callbacks=callbacks
)
# Save final model to MinIO
self.model.save('/tmp/final_model.h5')
self.s3.upload_file(
'/tmp/final_model.h5',
self.bucket,
'models/tensorflow/final_model.h5'
)
return history
def save_model_artifacts(self, model_name, version):
"""Save TensorFlow model artifacts to MinIO"""
import tempfile
import shutil
# Create temporary directory
with tempfile.TemporaryDirectory() as temp_dir:
# Save model in SavedModel format
model_path = f"{temp_dir}/saved_model"
self.model.save(model_path)
# Create tar archive
archive_path = f"{temp_dir}/{model_name}_{version}.tar.gz"
shutil.make_archive(
archive_path.replace('.tar.gz', ''),
'gztar',
model_path
)
# Upload to MinIO
s3_key = f"models/tensorflow/{model_name}/{version}/model.tar.gz"
self.s3.upload_file(archive_path, self.bucket, s3_key)
# Save model metadata
metadata = {
'name': model_name,
'version': version,
'framework': 'tensorflow',
'format': 'savedmodel',
'created_at': tf.timestamp().numpy(),
'model_config': self.model.get_config(),
'input_shape': [int(dim) for dim in self.model.input_shape],
'output_shape': [int(dim) for dim in self.model.output_shape]
}
metadata_key = f"models/tensorflow/{model_name}/{version}/metadata.json"
self.s3.put_object(
Bucket=self.bucket,
Key=metadata_key,
Body=json.dumps(metadata, indent=2),
ContentType='application/json'
)
return s3_key
# Usage example
s3_client = boto3.client('s3', endpoint_url='http://minio:9000',
aws_access_key_id='minioadmin',
aws_secret_access_key='minioadmin')
# Create a simple CNN model
model = tf.keras.Sequential([
tf.keras.layers.Conv2D(32, 3, activation='relu', input_shape=(224, 224, 3)),
tf.keras.layers.MaxPooling2D(),
tf.keras.layers.Conv2D(64, 3, activation='relu'),
tf.keras.layers.MaxPooling2D(),
tf.keras.layers.Conv2D(64, 3, activation='relu'),
tf.keras.layers.Flatten(),
tf.keras.layers.Dense(64, activation='relu'),
tf.keras.layers.Dense(10, activation='softmax')
])
trainer = TensorFlowMinIOTrainer(model, s3_client, 'training-data')
# Train model
history = trainer.train_model('train/', 'val/', epochs=5)
# Save model artifacts
model_path = trainer.save_model_artifacts('image_classifier', 'v1.0')
print(f"Model saved to: {model_path}")
The TensorFlow integration demonstrates the same end-to-end pattern through a different framework: tf.data pipelines read and preprocess from MinIO, callbacks protect the training run, and SavedModel artifacts plus metadata are written back as portable objects. Keeping both the PyTorch and TensorFlow paths on the same bucket conventions means one data lake serves teams with different framework preferences, and the registry stores their artifacts in a uniform, versioned layout.
The image dataset implementation is a good study in how tf.data pipelines are meant to be composed.
create_image_dataset() begins by listing image paths once and wrapping them in
tf.data.Dataset.from_tensor_slices, then applies a mapping that downloads and preprocesses each image. The
mapping runs inside tf.py_function, a necessary bridge because the download logic is Pythonic and side-effecting
(it makes a boto3 call) and cannot be expressed as a native TensorFlow graph op. The critical additions are
num_parallel_calls=tf.data.AUTOTUNE on the map and prefetch(tf.data.AUTOTUNE) on the dataset, which let
TensorFlow schedule downloads and preprocessing in parallel with training and hide I/O latency behind GPU
computation. Without those two lines, the pipeline would fetch images serially and dramatically underutilize the
accelerator.
The example also contains a subtle correctness issue that is common in real tf.data code.
load_and_preprocess_image() calls path.numpy().decode() and self.s3.get_object inside the mapping function,
which works because tf.py_function executes eagerly. However, tf.py_function exchanges tensors with Python by
copying data in and out of the graph, which adds overhead per sample; for production throughput, alternatives like
pre-downloading to local disk, using TensorFlow’s native tf.io.decode_jpeg on bytes already fetched in bulk, or
switching to a library like tf.io.gfile that integrates with object storage are often faster. The version shown
is correct and educational, but it should be treated as a baseline to optimize rather than a final architecture.
The text dataset takes a fundamentally different construction path, and that contrast is instructive. It builds the
dataset from a Python generator via tf.data.Dataset.from_generator with explicit output_signature tensor specs
- required so TensorFlow knows the shapes and dtypes of a lazy, non-sliceable source. The generator reads
.txtand.jsonobjects from MinIO and yields(text, label)tuples, then a mapping applies tokenization and the pipeline batches and prefetches. This generator-based approach is the TensorFlow counterpart of the streaming dataset in the PyTorch section: it never materializes the full corpus in memory and is ideal for text corpora that exceed RAM.
Finally, the training and artifact-persistence code shows Keras at its most productive. train_model() compiles a
sparse-categorical model and attaches three callbacks that would be hand-rolled in most other frameworks:
ModelCheckpoint persists the best model by validation accuracy, EarlyStopping halts training when validation
loss plateaus, and ReduceLROnPlateau halves the learning rate on stagnation. save_model_artifacts() then
packages the trained model in SavedModel format - the portable, directory-based format that survives across
TensorFlow versions - tars it for storage, uploads it under models/tensorflow/{name}/{version}/, and writes a
metadata JSON recording input/output shapes, config, and creation time. That metadata is exactly what the inference
engine’s _tensorflow_inference path will need when the model is promoted to serving.
Inference Patterns with MinIO
Real-Time Inference Architecture
Production inference systems require low-latency access to models, features, and cached predictions. MinIO’s high-performance object storage enables efficient inference patterns that scale with demand.
The MinIOInferenceEngine below applies a layered caching strategy to keep latency low while staying correct.
Models are loaded once and held in an in-memory model_cache, so repeated predictions never touch the object store
after the first fetch. Predictions themselves are cached behind a hash of the input plus model and version, first
in Redis when available and persistently in the inference-cache bucket - Redis gives single-digit-millisecond
hits, while the MinIO copy survives restarts and acts as a fallback if Redis is down. This hybrid pattern is the
standard answer to the cache-coherency question: the hot cache is volatile and fast, the durable cache is slow but
always present.
The engine dispatches to framework-specific inference paths (_pytorch_inference, _tensorflow_inference,
_sklearn_inference) based on the registry metadata, demonstrating why the metadata.json written at registration
time pays off - the serving layer can reconstruct and run any framework from the same bucket.
BatchInferenceProcessor then shows the offline counterpart: it walks a prefix of input files, runs predictions in
fixed-size batches, writes per-batch result JSON to an output bucket, and finishes with a summary document for
monitoring. Finally, ModelABTester implements online experimentation by routing traffic between two model
versions using a hash of the user ID, which keeps the same user on the same variant while respecting a configured
traffic split.
The architecture deliberately keeps models, features, and cached predictions in the same S3 namespace so that one set of permissions governs the entire inference path. The trade-off is that object storage cannot beat a locally mounted model for the absolute lowest latency; the in-memory and Redis caches exist precisely to bridge that gap, and the pattern scales horizontally because MinIO is a shared, lock-free source of truth.
import boto3
import torch
import json
import numpy as np
from datetime import datetime, timedelta
import redis
import hashlib
from typing import Dict, Any, Optional
import asyncio
import aiohttp
class MinIOInferenceEngine:
def __init__(self, s3_client, model_bucket='model-registry',
cache_bucket='inference-cache'):
self.s3 = s3_client
self.model_bucket = model_bucket
self.cache_bucket = cache_bucket
self.model_cache = {}
self.feature_cache = {}
# Redis for hot cache (optional)
try:
self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
except:
self.redis_client = None
def load_model(self, model_name, version='latest', cache=True):
"""Load model for inference with caching"""
cache_key = f"{model_name}_{version}"
if cache and cache_key in self.model_cache:
return self.model_cache[cache_key]
# Download model from MinIO
model_path = f"models/{model_name}/{version}"
try:
# Load model metadata
metadata_key = f"{model_path}/metadata.json"
metadata_obj = self.s3.get_object(Bucket=self.model_bucket, Key=metadata_key)
metadata = json.loads(metadata_obj['Body'].read())
# Load model artifacts
artifacts = {}
for artifact_name, artifact_path in metadata['artifacts'].items():
artifact_key = artifact_path.replace(f"s3://{self.model_bucket}/", "")
local_path = f"/tmp/{artifact_name}"
self.s3.download_file(self.model_bucket, artifact_key, local_path)
if artifact_name.endswith('.pt'):
artifacts[artifact_name] = torch.load(local_path, map_location='cpu')
elif artifact_name.endswith('.json'):
with open(local_path, 'r') as f:
artifacts[artifact_name] = json.load(f)
model_data = {
'metadata': metadata,
'artifacts': artifacts,
'loaded_at': datetime.now()
}
if cache:
self.model_cache[cache_key] = model_data
return model_data
except Exception as e:
raise RuntimeError(f"Failed to load model {model_name} v{version}: {e}")
def predict(self, model_name, input_data, version='latest',
use_cache=True, cache_ttl=3600):
"""Make prediction with caching"""
# Generate cache key for input
input_hash = hashlib.md5(str(input_data).encode()).hexdigest()
cache_key = f"prediction:{model_name}:{version}:{input_hash}"
# Check cache first
if use_cache:
cached_result = self._get_cached_prediction(cache_key)
if cached_result:
return cached_result
# Load model
model_data = self.load_model(model_name, version)
# Make prediction
try:
prediction = self._run_inference(model_data, input_data)
# Cache result
if use_cache:
self._cache_prediction(cache_key, prediction, cache_ttl)
return prediction
except Exception as e:
raise RuntimeError(f"Prediction failed: {e}")
def _run_inference(self, model_data, input_data):
"""Run actual inference based on model type"""
framework = model_data['metadata'].get('framework', 'unknown')
if framework == 'pytorch':
return self._pytorch_inference(model_data, input_data)
elif framework == 'tensorflow':
return self._tensorflow_inference(model_data, input_data)
elif framework == 'sklearn':
return self._sklearn_inference(model_data, input_data)
else:
raise ValueError(f"Unsupported framework: {framework}")
def _pytorch_inference(self, model_data, input_data):
"""PyTorch model inference"""
model_state = model_data['artifacts']['model.pt']
# Reconstruct model (simplified - in practice, you'd store architecture)
# This assumes you have a way to reconstruct the model architecture
model = self._reconstruct_pytorch_model(model_data['metadata'])
model.load_state_dict(model_state)
model.eval()
# Convert input to tensor
if isinstance(input_data, np.ndarray):
input_tensor = torch.from_numpy(input_data).float()
elif isinstance(input_data, list):
input_tensor = torch.tensor(input_data).float()
else:
input_tensor = input_data
# Add batch dimension if needed
if len(input_tensor.shape) == 1:
input_tensor = input_tensor.unsqueeze(0)
# Run inference
with torch.no_grad():
output = model(input_tensor)
return {
'predictions': output.numpy().tolist(),
'model_name': model_data['metadata']['name'],
'version': model_data['metadata']['version'],
'timestamp': datetime.now().isoformat()
}
def _get_cached_prediction(self, cache_key):
"""Get prediction from cache"""
try:
if self.redis_client:
cached = self.redis_client.get(cache_key)
if cached:
return json.loads(cached)
# Fallback to MinIO cache
try:
obj = self.s3.get_object(Bucket=self.cache_bucket, Key=f"predictions/{cache_key}.json")
cached_data = json.loads(obj['Body'].read())
# Check if cache is still valid
cached_time = datetime.fromisoformat(cached_data['cached_at'])
if datetime.now() - cached_time < timedelta(hours=1):
return cached_data['prediction']
except:
pass
except Exception as e:
print(f"Cache retrieval error: {e}")
return None
def _cache_prediction(self, cache_key, prediction, ttl):
"""Cache prediction result"""
try:
cache_data = {
'prediction': prediction,
'cached_at': datetime.now().isoformat(),
'ttl': ttl
}
if self.redis_client:
self.redis_client.setex(
cache_key,
ttl,
json.dumps(cache_data)
)
# Also cache in MinIO for persistence
self.s3.put_object(
Bucket=self.cache_bucket,
Key=f"predictions/{cache_key}.json",
Body=json.dumps(cache_data, indent=2),
ContentType='application/json'
)
except Exception as e:
print(f"Cache storage error: {e}")
class BatchInferenceProcessor:
def __init__(self, s3_client, inference_engine):
self.s3 = s3_client
self.inference_engine = inference_engine
def process_batch(self, input_bucket, input_prefix, output_bucket,
output_prefix, model_name, batch_size=100):
"""Process batch inference on data stored in MinIO"""
# List input files
response = self.s3.list_objects_v2(
Bucket=input_bucket,
Prefix=input_prefix
)
input_files = [obj['Key'] for obj in response.get('Contents', [])]
results = []
batch_count = 0
for i in range(0, len(input_files), batch_size):
batch_files = input_files[i:i + batch_size]
batch_results = []
for file_key in batch_files:
try:
# Download and process file
obj = self.s3.get_object(Bucket=input_bucket, Key=file_key)
if file_key.endswith('.json'):
input_data = json.loads(obj['Body'].read())
elif file_key.endswith('.csv'):
import pandas as pd
df = pd.read_csv(obj['Body'])
input_data = df.to_dict('records')
else:
# Handle other formats
input_data = obj['Body'].read()
# Make prediction
prediction = self.inference_engine.predict(
model_name, input_data, use_cache=False
)
batch_results.append({
'input_file': file_key,
'prediction': prediction,
'processed_at': datetime.now().isoformat()
})
except Exception as e:
batch_results.append({
'input_file': file_key,
'error': str(e),
'processed_at': datetime.now().isoformat()
})
# Save batch results
batch_output_key = f"{output_prefix}/batch_{batch_count:04d}.json"
self.s3.put_object(
Bucket=output_bucket,
Key=batch_output_key,
Body=json.dumps(batch_results, indent=2),
ContentType='application/json'
)
results.extend(batch_results)
batch_count += 1
print(f"Processed batch {batch_count}, files: {len(batch_files)}")
# Save summary
summary = {
'total_files': len(input_files),
'successful_predictions': len([r for r in results if 'prediction' in r]),
'failed_predictions': len([r for r in results if 'error' in r]),
'processing_time': datetime.now().isoformat(),
'model_used': model_name
}
summary_key = f"{output_prefix}/summary.json"
self.s3.put_object(
Bucket=output_bucket,
Key=summary_key,
Body=json.dumps(summary, indent=2),
ContentType='application/json'
)
return summary
# A/B Testing for Model Inference
class ModelABTester:
def __init__(self, inference_engine):
self.inference_engine = inference_engine
self.test_configs = {}
def setup_ab_test(self, test_name, model_a, model_b, traffic_split=0.5):
"""Setup A/B test between two models"""
self.test_configs[test_name] = {
'model_a': model_a,
'model_b': model_b,
'traffic_split': traffic_split,
'results': {'a': [], 'b': []}
}
def predict_with_ab_test(self, test_name, input_data, user_id=None):
"""Make prediction using A/B test configuration"""
if test_name not in self.test_configs:
raise ValueError(f"A/B test {test_name} not configured")
config = self.test_configs[test_name]
# Determine which model to use
if user_id:
# Consistent assignment based on user ID
import hashlib
hash_val = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16)
use_model_a = (hash_val % 100) < (config['traffic_split'] * 100)
else:
# Random assignment
import random
use_model_a = random.random() < config['traffic_split']
model_name = config['model_a'] if use_model_a else config['model_b']
variant = 'a' if use_model_a else 'b'
# Make prediction
prediction = self.inference_engine.predict(model_name, input_data)
# Track result
config['results'][variant].append({
'prediction': prediction,
'timestamp': datetime.now().isoformat(),
'user_id': user_id
})
return {
'prediction': prediction,
'variant': variant,
'model_used': model_name
}
# Usage examples
s3_client = boto3.client('s3', endpoint_url='http://minio:9000',
aws_access_key_id='minioadmin',
aws_secret_access_key='minioadmin')
# Setup inference engine
inference_engine = MinIOInferenceEngine(s3_client)
# Real-time prediction
input_data = [1.2, 3.4, 5.6, 7.8] # Example feature vector
result = inference_engine.predict('housing_price_predictor', input_data)
print(f"Prediction: {result}")
# Batch processing
batch_processor = BatchInferenceProcessor(s3_client, inference_engine)
summary = batch_processor.process_batch(
input_bucket='inference-input',
input_prefix='batch_001/',
output_bucket='inference-output',
output_prefix='results/batch_001',
model_name='housing_price_predictor'
)
print(f"Batch processing summary: {summary}")
# A/B testing
ab_tester = ModelABTester(inference_engine)
ab_tester.setup_ab_test('price_model_test', 'price_model_v1', 'price_model_v2', 0.3)
ab_result = ab_tester.predict_with_ab_test('price_model_test', input_data, user_id='user123')
print(f"A/B test result: {ab_result}")
The inference patterns show that low latency does not come from MinIO alone but from the layers around it: an in-memory model cache, a Redis hot cache for predictions, and a durable S3 cache for fallback. This tiered design keeps the object store as the source of truth while moving the hot path into memory, and it scales horizontally because every worker shares the same bucket namespace. When you deploy, run the A/B tester early - measuring model versions against real traffic is what justifies promoting one over another in the registry.
Tracing the predict() path shows how carefully the layers are ordered. The method first hashes the input and
consults the cache; only a cache miss falls through to load_model(), which in turn checks the in-memory
model_cache before touching MinIO. This ordering means the steady-state serving path - cache hit - never performs
a network I/O for either the model or the prediction, which is precisely the property that makes S3-based inference
viable at production latencies. The cache key includes model name, version, and an input hash, so different inputs
or different model versions never collide, and the TTL on cached predictions bounds staleness for time-varying
features.
The dual cache write in _cache_prediction() is the linchpin of the resilience story. When Redis is available, the
prediction is stored with setex (set with expiry) for single-digit-millisecond reads; the same result is also
persisted to the inference-cache bucket with a TTL embedded in the payload. _get_cached_prediction() reads
Redis first and falls back to MinIO when Redis is absent or empty, validating that a MinIO hit is not older than
one hour. The consequence is that a Redis crash degrades performance but never correctness - the durable S3 cache
absorbs the traffic while the cache warms again. Teams that skip the durable tier accept that every cache miss
after a Redis restart hits the model instead of the cache, which can spike latency across the fleet.
The framework dispatch in _run_inference() is the payoff of the registry design established earlier. It reads the
framework field from the model’s metadata and routes to _pytorch_inference, _tensorflow_inference, or
_sklearn_inference, each of which reconstructs the model from the stored artifacts. The PyTorch branch loads
state dicts, rebuilds the architecture (the code notes that real systems must persist the architecture itself, not
just the weights), sets eval() mode, and converts the input to a batch-shaped tensor before running under
torch.no_grad(). Note that this serving path returns the raw model output as a plain Python list - deliberately
framework-agnostic so that the API layer above can serialize it uniformly.
The BatchInferenceProcessor and ModelABTester classes round out the operational story. The batch processor
reads input files from one bucket prefix, chunks them into fixed-size batches, writes one result JSON per batch,
and finishes with a summary document - a pattern that makes partial results inspectable and resumable even if a run
dies halfway. The A/B tester uses a hash of the user ID to deterministically route each user to variant A or B
according to the configured split, which keeps each user on a consistent model across requests (vital for
statistically valid experiments) while still allowing an overall traffic ratio. Both classes reuse the same
inference engine, so batch scoring, online prediction, and experimentation all share one model-loading and caching
path.
Vector Database Integration
Storing Embeddings
Semantic search, recommendation systems, and RAG pipelines all depend on vector embeddings, and object storage is a
natural home for them even when a dedicated vector database handles similarity search. The store_embeddings()
function below writes each embedding as a JSON object keyed by the item’s ID, pairing the vector (serialized from
NumPy to a list) with its metadata and a creation timestamp. Because each object is self-contained, MinIO serves as
the durable source of truth: you can rebuild any vector index, re-embed for a new model version, or audit the
provenance of any vector by reading its object.
This storage pattern fits a two-tier architecture that is increasingly common in 2026. A vector database such as Milvus, Qdrant, or pgvector holds the HNSW or IVF indexes that make approximate nearest-neighbor search fast, while MinIO keeps the authoritative copy of the raw vectors and the source documents. When a model’s embedding dimension or version changes, you re-embed from the bucket rather than risking index corruption, then re-index the vector store - the object lake guarantees reproducibility, and the vector store guarantees query speed.
The trade-off of storing embeddings as individual JSON objects is that listing and bulk reads are slower than a
compact binary format. For large corpora, prefer batched Parquet files (one per embedding model and version) for
bulk export, while retaining the per-ID JSON layout for point lookups and metadata queries. Both layouts coexist
comfortably in MinIO, and the same bucket can hold vectors/, parquet/, and index snapshots without conflict.
import boto3
import json
import numpy as np
def store_embeddings(embeddings, metadata, bucket='embeddings'):
"""Store vector embeddings with metadata"""
s3 = boto3.client('s3',
endpoint_url='http://minio:9000',
aws_access_key_id='minioadmin',
aws_secret_access_key='minioadmin'
)
for i, (embedding, meta) in enumerate(zip(embeddings, metadata)):
key = f"vectors/{meta['id']}.json"
data = {
'id': meta['id'],
'embedding': embedding.tolist(),
'metadata': meta,
'created_at': datetime.now().isoformat()
}
s3.put_object(
Bucket=bucket,
Key=key,
Body=json.dumps(data),
ContentType='application/json'
)
# Store embeddings
embeddings = [np.random.randn(512) for _ in range(100)]
metadata = [{'id': f'item_{i}', 'category': 'test'} for i in range(100)]
store_embeddings(embeddings, metadata)
The embedding layout gives you durability, auditability, and reproducibility without committing to a single vector index technology. Because MinIO keeps the authoritative vectors, you can swap or rebuild your nearest-neighbor index - Milvus, Qdrant, pgvector, or FAISS - whenever the model or corpus changes, and the object lake doubles as the cold storage tier that keeps vector database costs in check.
The store_embeddings() function is a minimal but complete illustration of the pattern. It takes a list of NumPy
vectors and a matching list of metadata dicts, and for each pair it builds a JSON object containing the ID, the
vector converted to a plain list, the full metadata, and a creation timestamp. Storing the vector as a Python list
in JSON is deliberately simple - it is trivially inspectable and portable - but it costs roughly 50% more space
than a raw binary representation because JSON encodes each float as text. For real workloads, the same function
would serialize vectors with a compact format such as NumPy’s .npy bytes or a Parquet column, and reserve the
JSON layout for small metadata objects.
The important design property is that every object is self-contained: the vector, its ID, its metadata, and its timestamp all live in one file. That self-containment is what makes the two-tier architecture work. When you need to rebuild a vector index, you read the authoritative vectors from MinIO; when you need to audit why a particular item ranks the way it does, you open its object and see the exact vector and metadata that produced the ranking; when you upgrade the embedding model, you re-embed from the stored source material and write new vectors alongside the old ones under a versioned prefix. This is the same philosophy that runs through the whole guide - MinIO is the durable, queryable record, and any specialized service built on top is rebuildable from it.
There is also a clear scaling path from this minimal function to a production embedding store. The per-object loop shown here becomes slow beyond tens of thousands of objects, so production versions typically list all metadata keys first, then parallelize the writes with a thread pool (exactly as the ingestion class demonstrated), or write large batches as single Parquet files partitioned by embedding model and version. The retention story is equally straightforward: since embeddings for an old model version are ordinary objects, you can archive them with MinIO lifecycle rules or simply leave them in place for reproducibility. Nothing about the vector layout locks you into a specific vendor, and that portability is the main reason to keep the authoritative copy in MinIO rather than only inside a vector database.
Complete ML Pipeline
End-to-End Example
The end-to-end example below assembles every component from this guide into a single orchestrated flow.
MLPipeline initializes a MinIO client and an output bucket, then defines the five lifecycle stages - ingest,
preprocess, train, evaluate, and deploy - that run() executes in sequence. The method bodies are stubbed here
with pass because each stage is precisely the machinery implemented earlier: ingestion maps to the
HighThroughputIngestion class, preprocessing to the Spark pipeline and data validator, training to the PyTorch or
TensorFlow trainers, and deployment to the MLModelRegistry and inference engine.
Structuring the pipeline as a class with discrete methods, rather than a linear script, gives you three practical
benefits. First, each stage can be run and debugged independently - you can re-run train() alone after improving
the model without re-ingesting data. Second, the methods are natural seams for observability: wrap each one to emit
lineage records, duration metrics, and model metrics that land back in MinIO. Third, the interface stays stable
while the implementations evolve, so you can swap a scikit-learn model for a PyTorch one, or a Spark job for a Ray
job, without rewriting the orchestration.
This pattern is intentionally minimal compared to a full workflow scheduler such as Airflow, Prefect, or Argo
Workflows, which add retries, backfills, and DAG visualization. For many teams the plain class is enough, and it
composes cleanly with a scheduler later: the scheduler calls pipeline.run() (or individual stages) as tasks,
while MinIO remains the single source of truth for every artifact the pipeline touches.
class MLPipeline:
def __init__(self, minio_endpoint, bucket_prefix='ml-pipeline'):
self.s3 = boto3.client('s3',
endpoint_url=minio_endpoint,
aws_access_key_id='minioadmin',
aws_secret_access_key='minioadmin'
)
self.bucket = bucket_prefix
self._ensure_bucket()
def _ensure_bucket(self):
try:
self.s3.create_bucket(Bucket=self.bucket)
except:
pass
def ingest(self, source_path):
"""Ingest raw data"""
# Upload raw data
pass
def preprocess(self):
"""Clean and transform data"""
# Process data
pass
def train(self):
"""Train model"""
# Training loop
pass
def evaluate(self):
"""Evaluate model"""
# Evaluation
pass
def deploy(self):
"""Deploy to inference"""
# Save model and metadata
pass
def run(self):
"""Run complete pipeline"""
self.ingest()
self.preprocess()
self.train()
self.evaluate()
self.deploy()
# Run pipeline
pipeline = MLPipeline('http://minio:9000')
pipeline.run()
The complete pipeline demonstrates the central theme of this guide: MinIO is not just where AI data lives but the connective tissue of the whole ML lifecycle. Raw files enter one bucket, transformed data exits another, features and models are versioned, and inference reads from the same namespace - all behind a single S3-compatible API. Start with the bucket layout and this pipeline skeleton, then deepen each stage with the production classes from the sections above.
Even in this stub form, the skeleton encodes real architectural decisions worth preserving. The class holds exactly
one shared resource - the S3 client - and constructs it once in the constructor, so every stage reuses the same
connection pool rather than creating a fresh client per call. The _ensure_bucket() call in __init__ repeats the
idempotent-bootstrap pattern from the data lake class, guaranteeing that the pipeline’s working bucket exists
before any stage runs. Each lifecycle method takes the arguments it needs and returns nothing in the stub, but the
signatures define the seams: ingest(source_path) knows where data comes from, and the other stages know that
whatever they need was left in MinIO by the stage before them.
The most valuable habit this example teaches is to build ML systems around explicit, separately executable stages
rather than one monolithic training script. Every stage boundary is a natural checkpoint: after ingest(), the raw
data exists in MinIO and can be inspected; after preprocess(), the training splits exist and can be validated;
after train(), the model artifacts exist in the registry and can be evaluated. If any stage fails, you fix it and
re-run just that stage, because the inputs it needs are already persisted. That recoverability is the practical
definition of a production ML pipeline, and it emerges almost for free from the object-storage-centric design shown
throughout this article.
A final observation about how far this architecture can scale. The pipeline class as written runs sequentially in a
single process, which is fine for a batch model that retrains nightly. But because every stage persists to MinIO,
the same skeleton scales up in two directions without redesign: horizontally, the stages can move to separate
worker instances (or an Airflow/Prefect DAG) and communicate purely through the buckets; and vertically, each stage
can be upgraded to the production class it represents - the ingestion stage to HighThroughputIngestion,
preprocessing to the Spark pipeline plus validator, training to PyTorchMinIOTrainer or TensorFlowMinIOTrainer,
and deployment to MLModelRegistry and MinIOInferenceEngine. The bucket layout and the S3 contract are the
constant; the implementations behind each method are what grow.
Conclusion
MinIO provides ideal storage for AI/ML workloads. Key capabilities: high-throughput data ingestion for training, efficient model checkpoint storage, feature store backends, vector embedding storage, and seamless integration with ML frameworks. Using MinIO as your AI data lake simplifies infrastructure while maintaining S3 compatibility.
In the final article, we’ll explore real-world MinIO use cases.
Comments