Enterprise ML / Data Stack

Full Implementation Plan — Docker Swarm (3× GPU Nodes)

Environment Approach
Docker Swarm · 3 GPU Nodes Phased rollout, 16 layers

Existing Stack Technologies

MinIO · MLflow · Airflow · Prefect · 21 new services added
CVAT
Overview & Architecture
This document provides a complete, phased implementation plan for deploying an enterprise-grade ML/Data
stack on your existing Docker Swarm cluster. Each section covers prerequisites, deployment configuration,
integration points, and validation steps.

Cluster Topology
node-1 (manager) Swarm manager. Runs control-plane services: Airflow scheduler, Prefect server,
Temporal server, metadata DBs, Vault, Keycloak, OpenMetadata.

node-2 (worker) GPU worker. Primary training node. Hosts Ray head node, JupyterHub, Triton
Inference Server.

node-3 (worker) GPU worker. Secondary compute. Ray worker, BentoML, Evidently, ClickHouse.

Shared (all nodes) MinIO (already running), Prometheus node exporters, cAdvisor, NVIDIA DCGM
exporter.

Network Design
Create dedicated Docker overlay networks to isolate traffic concerns:
• ml-data-net — data pipeline traffic (Kafka, Spark, Flink, Delta Lake reads/writes to MinIO)
• ml-train-net — training traffic (Ray, JupyterHub, MLflow, Feast)
• ml-serve-net — inference traffic (Triton, BentoML, Ray Serve, Redis)
• ml-ops-net — control plane (Airflow, Prefect, Temporal, Vault, Keycloak)
• ml-obs-net — observability (Prometheus, Grafana, cAdvisor, DCGM)

⚠ All services must join ml-ops-net for Vault secret injection. Services needing cross-network communication use
attachable overlay networks.
 1 Phase 1 — Foundation (Week 1)

Deploy the security and platform foundation before any application services. Everything else depends on these.

1.1 — HashiCorp Vault

Purpose
Central secrets management. Every service fetches credentials at runtime — no secrets in Compose files,
environment variables, or Git.

Docker Swarm Stack

# [Link]
version: "3.8"
services:
vault:
image: hashicorp/vault:1.16
command: server
environment:
VAULT_LOCAL_CONFIG: |
storage "file" { path = "/vault/data" }
listener "tcp" {
address = "[Link]:8200"
tls_disable = true # use Traefik TLS termination
}
ui = true
api_addr = "[Link]
ports:
- "8200:8200"
volumes:
- vault_data:/vault/data
cap_add:
- IPC_LOCK
networks:
- ml-ops-net
deploy:
placement:
constraints: [[Link] == manager]

volumes:
vault_data:
networks:
ml-ops-net:
external: true

Initialization Steps
1. Deploy: docker stack deploy -c [Link] vault
2. Initialize: docker exec <vault_container> vault operator init -key-shares=5 -key-threshold=3
3. Save all 5 unseal keys and the root token in a secure offline location
4. Unseal with any 3 keys: vault operator unseal <key>
5. Enable secrets engine: vault secrets enable -path=ml-stack kv-v2
6. Create a policy per service (example for Airflow): vault policy write airflow /policies/[Link]

Secret Paths Convention

ml-stack/databases/postgres PostgreSQL passwords for all services

ml-stack/minio MinIO access key and secret key

 ml-stack/mlflow MLflow tracking server credentials

ml-stack/feast Feast registry and online store credentials

ml-stack/kafka Kafka SASL credentials and SSL certs

ml-stack/clickhouse ClickHouse admin and service users

📝 Use Vault Agent Sidecar or envconsul to inject secrets into containers at startup. Avoid passing Vault tokens directly
to app containers.

1.2 — PostgreSQL (HA Metadata Store)

Purpose
Central relational metadata store for Airflow, Prefect, Temporal, MLflow, Feast registry, and OpenMetadata.

Deployment
# [Link]
version: "3.8"
services:
postgres:
image: postgres:16
environment:
POSTGRES_PASSWORD_FILE: /run/secrets/pg_password
volumes:
- postgres_data:/var/lib/postgresql/data
- ./[Link]:/docker-entrypoint-initdb.d/[Link]
networks: [ml-ops-net]
deploy:
placement:
constraints: [[Link] == manager]
secrets:
- pg_password

secrets:
pg_password:
external: true # stored in Docker Swarm secrets, sourced from Vault

volumes:
postgres_data:

[Link] — create all required databases

CREATE DATABASE airflow;
CREATE DATABASE prefect;
CREATE DATABASE temporal;
CREATE DATABASE mlflow;
CREATE DATABASE feast_registry;
CREATE DATABASE openmetadata;
CREATE DATABASE metabase;

-- Create dedicated service users

CREATE USER airflow_user WITH PASSWORD '<from-vault>';
GRANT ALL PRIVILEGES ON DATABASE airflow TO airflow_user;
-- Repeat for each service...
1.3 — Redis
Purpose
Shared cache, Airflow Celery broker, Feast online store backend, Ray task state cache.
# [Link]
services:
redis:
image: redis:7-alpine
command: redis-server --requirepass ${REDIS_PASSWORD} --maxmemory 4gb --maxmemory-policy
allkeys-lru
volumes:
- redis_data:/data
networks: [ml-ops-net, ml-serve-net]
deploy:
placement:
constraints: [[Link] == manager]
 2 Phase 2 — Storage & Lakehouse (Week 2)

Build the lakehouse foundation on top of your existing MinIO. Delta Lake provides ACID transactions, schema
evolution, and time travel on object storage.

2.1 — Delta Lake on MinIO

What Delta Lake adds to MinIO
ACID transactions Safe concurrent reads and writes — no partial writes or dirty reads

Time travel Query data as-of any previous version: SELECT * FROM table VERSION AS OF 10

Schema enforcement Reject writes that violate the table schema

Schema evolution ALTER TABLE to add/rename columns without rewriting data

Compaction (OPTIMIZE) Merge small Parquet files into larger ones for read performance

Z-ORDER clustering Co-locate related data in the same files for predicate pushdown

MinIO Configuration for Delta Lake

Delta Lake requires S3 path-style access and versioning. Configure MinIO buckets:
# Run with MinIO client (mc)
mc alias set minio [Link] <access-key> <secret-key>

# Create dedicated buckets per layer

mc mb minio/delta-raw # raw ingestion zone
mc mb minio/delta-bronze # cleaned/typed data
mc mb minio/delta-silver # joined/enriched data
mc mb minio/delta-gold # aggregated/ML-ready features
mc mb minio/delta-models # model artifacts (used by MLflow)
mc mb minio/delta-checkpoints # Spark and Flink checkpoints

# Enable versioning on delta buckets

mc version enable minio/delta-raw
mc version enable minio/delta-bronze
mc version enable minio/delta-silver
mc version enable minio/delta-gold

Spark Delta Lake Configuration

Delta Lake is used via Spark. When you deploy Spark (Phase 4), apply these settings:
# [Link]
[Link] [Link]:delta-spark_2.12:3.2.0,\
[Link]:hadoop-aws:3.3.4,\
[Link]:aws-java-sdk-bundle:1.12.262

[Link] [Link]
[Link].spark_catalog [Link]

# MinIO / S3A configuration

[Link] [Link]
[Link] <from-vault>
[Link] <from-vault>
[Link] true
[Link] [Link].s3a.S3AFileSystem

Creating Your First Delta Table

 # pyspark example
from delta import DeltaTable
from [Link] import SparkSession

spark = [Link] \
.appName("delta-init") \
.getOrCreate()

# Write a Delta table

[Link] \
.format("delta") \
.mode("overwrite") \
.save("s3a://delta-bronze/events")

# Time travel query

[Link] \
.format("delta") \
.option("versionAsOf", 5) \
.load("s3a://delta-bronze/events")

2.2 — LakeFS
Purpose
Git-like branching and versioning for your data lake. Create branches for experiments, merge changes after
validation, roll back bad data loads — all without copying data.

Docker Swarm Deployment

# [Link]
services:
lakefs:
image: treeverse/lakefs:1.25
command: run --config /etc/lakefs/[Link]
configs:
- source: lakefs_config
target: /etc/lakefs/[Link]
ports:
- "8080:8080"
networks: [ml-ops-net, ml-data-net]
deploy:
placement:
constraints: [[Link] == manager]

configs:
lakefs_config:
external: true # contains MinIO endpoint and DB connection

# lakefs [Link] content

database:
type: postgres
postgres:
connection_string: postgres://lakefs_user:<pw>@postgres:5432/lakefs

blockstore:
type: s3
s3:
force_path_style: true
endpoint: [Link]
credentials:
 access_key_id: <from-vault>
secret_access_key: <from-vault>

auth:
encrypt:
secret_key: <random-32-char-string-from-vault>

LakeFS Workflow Integration

New data branch lakectl branch create lakefs://my-repo/feature/new-dataset --source main

Write to branch Point Spark/PySpark at s3a://my-repo/feature/new-dataset/ instead of main

Validate & merge Run Great Expectations on branch, then lakectl merge feature/new-dataset main

Rollback bad load lakectl branch revert lakefs://my-repo/main --parent 1

Airflow integration Use LakeFSHook from lakefs-sdk-python in your DAGs

 3 Phase 3 — Streaming & Ingestion (Week 3)

3.1 — RabbitMQ
Purpose
Message broker for event-driven ingestion, decoupling producers from consumers, task queuing, and as the
Celery broker for Airflow (replacing Redis for that role if preferred).

Docker Swarm Deployment

# [Link]
services:
rabbitmq:
image: rabbitmq:3.13-management
hostname: rabbitmq
environment:
RABBITMQ_DEFAULT_USER_FILE: /run/secrets/rmq_user
RABBITMQ_DEFAULT_PASS_FILE: /run/secrets/rmq_pass
RABBITMQ_DEFAULT_VHOST: ml_stack
volumes:
- rabbitmq_data:/var/lib/rabbitmq
- ./[Link]:/etc/rabbitmq/[Link]
ports:
- "5672:5672" # AMQP
- "15672:15672" # Management UI
networks: [ml-data-net, ml-ops-net]
deploy:
replicas: 1
placement:
constraints: [[Link] == manager]

Exchange & Queue Architecture

Exchange: [Link] (topic) Routes incoming data events by routing key (e.g. sensor.#, [Link],
[Link])

Exchange: [Link] (fanout) Broadcasts system events to all subscribers (pipeline completions, alerts)

Queue: raw-data-ingest Bound to [Link] — consumed by Spark/Flink ingestion workers

Queue: airflow-tasks Celery queue for Airflow worker tasks (alternative to Redis broker)

Queue: prefect-tasks Prefect work pool queue for ML flow executions

Queue: model-retraining Triggers from monitoring drift detection to Prefect training flows

Python Producer Example

import pika, json

connection = [Link](
[Link](
host='rabbitmq',
virtual_host='ml_stack',
credentials=[Link]('user', 'pass')
)
)
channel = [Link]()
channel.exchange_declare(exchange='[Link]', exchange_type='topic', durable=True)

channel.basic_publish(
exchange='[Link]',
 routing_key='[Link]',
body=[Link]({"bucket": "delta-raw", "key": "images/batch_001.tar"}),
properties=[Link](delivery_mode=2) # persistent
)
[Link]()

Airflow Integration
Use RabbitMQ as Airflow's Celery broker by updating the Airflow Helm/Compose config:
# airflow environment
AIRFLOW__CELERY__BROKER_URL: amqp://user:pass@rabbitmq:5672/ml_stack
AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow_user:pass@postgres:5432/airflow
 4 Phase 4 — Batch Processing (Week 4)

4.1 — Apache Spark (PySpark)

Deployment: Spark Standalone on Docker Swarm
# [Link]
services:
spark-master:
image: bitnami/spark:3.5
environment:
- SPARK_MODE=master
- SPARK_MASTER_WEBUI_PORT=8090
ports:
- "7077:7077"
- "8090:8090"
volumes:
- ./[Link]:/opt/bitnami/spark/conf/[Link]
networks: [ml-data-net, ml-ops-net]
deploy:
placement:
constraints: [[Link] == manager]

spark-worker:
image: bitnami/spark:3.5
environment:
- SPARK_MODE=worker
- SPARK_MASTER_URL=spark://spark-master:7077
- SPARK_WORKER_MEMORY=16G
- SPARK_WORKER_CORES=8
networks: [ml-data-net]
deploy:
replicas: 3 # one per node
placement:
max_replicas_per_node: 1

Key PySpark + Delta Lake Patterns

# Ingest from RabbitMQ message → Delta Lake (bronze)
from [Link] import SparkSession
from [Link] import from_json, col, current_timestamp

spark = [Link]("ingest-bronze").getOrCreate()

# Read from S3 landing zone (files dropped by RabbitMQ consumer)

raw_df = [Link]("s3a://delta-raw/events/date=2025-03-12/")

# Apply schema, add metadata, write to bronze Delta table

bronze_df = raw_df \
.withColumn("ingested_at", current_timestamp()) \
.withColumn("source_file", col("_metadata.file_path"))

bronze_df.write \
.format("delta") \
.mode("append") \
.option("mergeSchema", "true") \
.partitionBy("date") \
.save("s3a://delta-bronze/events")

Airflow DAG for Spark Jobs

 from [Link].spark_submit import SparkSubmitOperator

spark_ingest = SparkSubmitOperator(
task_id="ingest_to_bronze",
application="/opt/airflow/jobs/ingest_bronze.py",
conn_id="spark_default", # points to spark://spark-master:7077
packages="[Link]:delta-spark_2.12:3.2.0",
conf={"[Link]": "[Link]"},
dag=dag
)

4.2 — Ray Cluster

Purpose
Distributed ML compute across your 3 GPU nodes. Handles distributed PyTorch/TF training, hyperparameter
tuning, and model serving. The primary bridge between your data layer and GPU hardware.

Docker Swarm Deployment

# [Link]
services:
ray-head:
image: rayproject/ray:2.20.0-gpu
command: ray start --head --port=6379 --dashboard-host=[Link] --num-gpus=0
environment:
- RAY_DISABLE_DOCKER_CPU_WARNING=1
ports:
- "6379:6379" # Redis / GCS
- "8265:8265" # Dashboard
- "10001:10001" # Client port
volumes:
- /tmp/ray:/tmp/ray
networks: [ml-train-net, ml-ops-net]
deploy:
placement:
constraints: [[Link] == manager]

ray-worker:
image: rayproject/ray:2.20.0-gpu
command: >
ray start --address=ray-head:6379
--num-cpus=16 --num-gpus=2
--block
volumes:
- /tmp/ray:/tmp/ray
runtime: nvidia
environment:
- NVIDIA_VISIBLE_DEVICES=all
networks: [ml-train-net]
deploy:
replicas: 2
placement:
max_replicas_per_node: 1
constraints: [[Link] == true]

Prefect + Ray Training Flow

from prefect import flow, task
from prefect_ray import RayTaskRunner
import ray
from ray import train
from [Link] import TorchTrainer

@flow(task_runner=RayTaskRunner(address="ray://ray-head:10001"))
def training_pipeline(experiment_name: str, config: dict):
preprocessed_data = preprocess_data(config)
model_uri = train_model(preprocessed_data, config, experiment_name)
evaluate_model(model_uri, experiment_name)
return model_uri

@task
def train_model(data_ref, config, experiment_name):
trainer = TorchTrainer(
train_loop_per_worker=train_func,
train_loop_config=config,
scaling_config=[Link](
num_workers=2, use_gpu=True, resources_per_worker={"GPU": 1}
),
)
result = [Link]()
return [Link]
 5 Phase 5 — Databases & Query (Week 5)

5.1 — ClickHouse
Purpose
Column-oriented OLAP database for sub-second analytical queries on large datasets. Use for: model monitoring
metrics, pipeline run analytics, event analytics, and dashboards (Metabase connects here).

Docker Swarm Deployment

# [Link]
services:
clickhouse:
image: clickhouse/clickhouse-server:24.3
ulimits:
nofile:
soft: 262144
hard: 262144
volumes:
- clickhouse_data:/var/lib/clickhouse
- ./[Link]:/etc/clickhouse-server/config.d/[Link]
- ./[Link]:/etc/clickhouse-server/users.d/[Link]
ports:
- "8123:8123" # HTTP interface
- "9000:9000" # Native TCP (avoid conflict with MinIO — remap MinIO to 9001)
networks: [ml-data-net, ml-obs-net]
deploy:
placement:
constraints: [[Link] == true]

Core Tables Setup

-- Model predictions log (for Evidently monitoring)
CREATE TABLE [Link] (
model_name LowCardinality(String),
model_version String,
prediction_id UUID DEFAULT generateUUIDv4(),
features String, -- JSON
prediction Float32,
ground_truth Nullable(Float32),
latency_ms Float32,
timestamp DateTime DEFAULT now()
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (model_name, timestamp);

-- Pipeline run metrics (feeds Metabase dashboards)

CREATE TABLE ops.pipeline_runs (
pipeline_name LowCardinality(String),
run_id String,
status LowCardinality(String),
started_at DateTime,
finished_at Nullable(DateTime),
rows_processed UInt64,
error_message Nullable(String)
) ENGINE = MergeTree()
ORDER BY (pipeline_name, started_at);

Integrations
Evidently AI Evidently pushes drift reports as JSON; a lightweight Python consumer writes rows
to ClickHouse

Metabase Connect Metabase to ClickHouse via JDBC/HTTP driver for dashboards

Airflow SparkSubmitOperator results logged to ClickHouse via airflow callbacks

Grafana ClickHouse Grafana plugin for operational dashboards alongside Prometheus

 6 Phase 6 — Feature Store (Week 6)

6.1 — Feast
Architecture
Feast bridges your offline (Delta Lake on MinIO) and online (Redis) stores, eliminating training/serving skew.
Offline store Delta Lake / Parquet on MinIO. Historical features for training.

Online store Redis. Low-latency feature lookup during inference (<5ms).

Registry PostgreSQL. Feature definitions, metadata, and lineage.

Materialization Feast job that syncs features from offline to online store on a schedule.

Docker Swarm Deployment

# [Link]
services:
feast-server:
image: feastdev/feature-server:0.40.0
command: serve --host [Link] --port 6566
volumes:
- ./feature_store.yaml:/feature_store/feature_store.yaml
ports:
- "6566:6566"
networks: [ml-train-net, ml-serve-net]
deploy:
placement:
constraints: [[Link] == true]

# feature_store.yaml
project: ml_stack
registry: postgresql://feast_user:<pw>@postgres:5432/feast_registry

provider: local

offline_store:
type: file # reads Parquet/Delta directly from MinIO via s3fs

online_store:
type: redis
connection_string: redis://:password@redis:6379

Feature Definition Example

# features/user_activity.py
from feast import Entity, Feature, FeatureView, FileSource, ValueType
from datetime import timedelta

user = Entity(name="user_id", value_type=ValueType.INT64)

user_activity_source = FileSource(
path="s3a://delta-gold/user_activity/",
timestamp_field="event_timestamp"
)

user_activity_fv = FeatureView(
name="user_activity",
entities=["user_id"],
 ttl=timedelta(days=30),
features=[
Feature(name="session_count_7d", dtype=ValueType.INT32),
Feature(name="avg_session_duration", dtype=[Link]),
Feature(name="last_active_days_ago", dtype=ValueType.INT32),
],
source=user_activity_source
)

Materialization DAG (Airflow)

# Run daily to sync offline → online store
from [Link] import PythonOperator

def materialize_features():
import subprocess
[Link]([
"feast", "-c", "/feature_store",
"materialize-incremental",
[Link]().isoformat()
], check=True)

materialize_task = PythonOperator(
task_id="feast_materialize",
python_callable=materialize_features,
dag=dag
)
 7 Phase 7 — Training & Experimentation (Week 7)

7.1 — JupyterHub
Purpose
Multi-user notebook environment with GPU access, connected to your full stack: MinIO, MLflow, Feast, Ray, and
Delta Lake — all accessible from notebooks.

Docker Swarm Deployment

# [Link]
services:
jupyterhub:
image: jupyterhub/jupyterhub:4.1
command: jupyterhub -f /srv/jupyterhub/jupyterhub_config.py
volumes:
- ./jupyterhub_config.py:/srv/jupyterhub/jupyterhub_config.py
- jupyterhub_data:/data
- /var/run/[Link]:/var/run/[Link] # spawns user containers
ports:
- "8000:8000"
networks: [ml-train-net, ml-ops-net]
deploy:
placement:
constraints: [[Link] == manager]

# jupyterhub_config.py
from dockerspawner import DockerSpawner

[Link].spawner_class = DockerSpawner
[Link] = "your-registry/ml-notebook:latest"

# Mount MinIO credentials and Feast config

[Link] = {
"MLFLOW_TRACKING_URI": "[Link]
"FEAST_CONFIG": "/home/jovyan/feature_store/feature_store.yaml"
}

# GPU access - assign GPU per user

[Link].extra_host_config = {
"device_requests": [{"count": 1, "capabilities": [["gpu"]]}]
}

# Auth via Keycloak (when deployed in Phase 9)

[Link].authenticator_class = "[Link]"

ML Notebook Docker Image

# Dockerfile for user notebook image
FROM jupyter/scipy-notebook:latest
USER root

RUN pip install \

delta-spark==3.2.0 \
mlflow==2.14 \
feast==0.40.0 \
ray[all]==2.20.0 \
evidently==0.4.0 \
torch torchvision \
pyspark==3.5.0 \
 s3fs boto3

# Pre-configure Spark for Delta Lake and MinIO

COPY [Link] /usr/local/spark/conf/
USER jovyan
 8 Phase 8 — Model Serving (Week 8)

8.1 — NVIDIA Triton Inference Server

Purpose
GPU-accelerated model serving with dynamic batching, multi-model support, TensorRT optimization, and model
ensembles. The production inference engine for your GPU nodes.

Docker Swarm Deployment

# [Link]
services:
triton:
image: [Link]/nvidia/tritonserver:24.05-py3
command: >
tritonserver
--model-repository=s3://delta-models/triton/
--model-control-mode=poll
--repository-poll-secs=30
--allow-metrics=true
--metrics-port=8002
environment:
AWS_ACCESS_KEY_ID: <from-vault>
AWS_SECRET_ACCESS_KEY: <from-vault>
AWS_DEFAULT_REGION: us-east-1
S3_ENDPOINT_URL: [Link]
ports:
- "8001:8001" # HTTP REST
- "8003:8003" # gRPC
- "8002:8002" # Metrics (Prometheus)
runtime: nvidia
environment:
NVIDIA_VISIBLE_DEVICES: all
networks: [ml-serve-net, ml-obs-net]
deploy:
placement:
constraints: [[Link] == true]

Model Repository Structure in MinIO

s3://delta-models/triton/
├── resnet50_classifier/
│ ├── [Link]
│ └── 1/
│ └── [Link] # or [Link] (TensorRT)
├── text_embedder/
│ ├── [Link]
│ └── 1/
│ └── [Link]
└── ensemble_pipeline/ # multi-model pipeline
├── [Link]
└── 1/

# [Link] example
name: "resnet50_classifier"
platform: "onnxruntime_onnx"
max_batch_size: 64
dynamic_batching {
preferred_batch_size: [8, 16, 32]
max_queue_delay_microseconds: 5000
}
 input [{ name: "input", data_type: TYPE_FP32, dims: [3, 224, 224] }]
output [{ name: "output", data_type: TYPE_FP32, dims: [1000] }]

MLflow → Triton Deployment Pipeline

# BentoML handles the MLflow → Triton packaging
import bentoml
import mlflow

# 1. Pull model from MLflow registry

mlflow_model = [Link].load_model("models:/my_model/Production")

# 2. Save to BentoML
bento_model = [Link].save_model("my_model", mlflow_model)

# 3. Export to ONNX for Triton

[Link].save_model("my_model_onnx", onnx_model)

# 4. Push to MinIO model repository

# (script copies to s3://delta-models/triton/my_model/1/[Link])

8.2 — BentoML
Purpose
Python-first model packaging, preprocessing/postprocessing logic, and serving abstraction. Sits in front of Triton
for models requiring Python business logic, or serves simpler models independently.

Docker Swarm Deployment

# [Link]
services:
bentoml:
image: your-registry/bentoml-service:latest # built by CI/CD
ports:
- "3000:3000" # HTTP
environment:
BENTOML_HOME: /bentoml
TRITON_HOST: triton
TRITON_PORT: "8001"
networks: [ml-serve-net]
deploy:
replicas: 2
placement:
constraints: [[Link] == true]

# [Link] — BentoML service with Triton backend

import bentoml
import numpy as np
from [Link] import NumpyNdarray, JSON
import [Link] as httpclient

svc = [Link]("image_classifier")

@[Link](input=NumpyNdarray(), output=JSON())
def predict(input_data: [Link]):
# Preprocessing
processed = preprocess(input_data)
# Forward to Triton
client = [Link]("triton:8001")
inputs = [[Link]("input", [Link], "FP32")]
inputs[0].set_data_from_numpy(processed)
response = [Link]("resnet50_classifier", inputs)

# Postprocessing
logits = response.as_numpy("output")
return {"class": int([Link](logits)), "confidence": float([Link](logits))}
 9 Phase 9 — Monitoring & Quality (Week 9)

9.1 — Evidently AI
Purpose
Model and data monitoring. Detects data drift, prediction drift, and data quality issues in production. Reports feed
into ClickHouse and trigger retraining via RabbitMQ.

Deployment & Integration Pattern

# [Link]
services:
evidently-ui:
image: evidently/evidently-service:latest
ports:
- "8085:8085"
volumes:
- evidently_data:/app/workspace
networks: [ml-obs-net, ml-serve-net]

# monitoring_job.py — runs as a Prefect flow (daily)

from prefect import flow, task
from [Link] import Report
from evidently.metric_preset import DataDriftPreset, ModelPerformancePreset
from [Link] import *
import pandas as pd
import clickhouse_driver

@flow
def run_model_monitoring(model_name: str, date: str):
report = check_drift(model_name, date)
store_to_clickhouse(report, model_name, date)
if report["drift_detected"]:
trigger_retraining(model_name)

@task
def check_drift(model_name: str, date: str):
# Pull reference data (training distribution)
ref_df = pd.read_parquet(f"s3a://delta-gold/reference/{model_name}/")

# Pull production predictions from ClickHouse

client = clickhouse_driver.Client("clickhouse")
prod_data = client.query_dataframe(
f"SELECT * FROM [Link] WHERE model_name='{model_name}' AND
toDate(timestamp)='{date}'"
)

report = Report(metrics=[DataDriftPreset(), DataQualityPreset()])

[Link](reference_data=ref_df, current_data=prod_data)

return report.as_dict()

@task
def trigger_retraining(model_name: str):
import pika
# Publish to RabbitMQ model-retraining queue
# ... (see RabbitMQ section for pattern)

Evidently + Grafana Dashboard

Drift score over time Evidently writes drift_score per feature to ClickHouse; Grafana visualizes trend

Missing value rate Data quality metrics tracked per batch in ClickHouse

Prediction distribution Histogram of model outputs over time — spot distribution shift visually

Retraining trigger log Table of when drift thresholds were crossed and retraining was triggered
 10 Phase 10 — Orchestration Layer (Week 10)

You already have Airflow and Prefect. This phase adds Temporal for durable long-running workflows and
completes the integration wiring between all three.

10.1 — Temporal
When to use Temporal vs Airflow vs Prefect
Temporal Long-running workflows (hours/days), human-in-the-loop (annotation review →
approve → retrain), durable retries across failures, async operations. Workflow state
survives cluster restarts.

Prefect ML training flows, dynamic pipelines, Ray job submission, hyperparameter sweeps,
model evaluation. Python-native, fast iteration.

Airflow Scheduled batch ETL, Spark job submission, dbt runs, fixed-schedule SLA-bound
pipelines, Great Expectations checkpoint runs.

Docker Swarm Deployment

# [Link]
services:
temporal:
image: temporalio/auto-setup:1.24
environment:
DB: postgresql
DB_PORT: 5432
POSTGRES_USER: temporal_user
POSTGRES_PWD_FILE: /run/secrets/temporal_pg_pass
POSTGRES_SEEDS: postgres
DYNAMIC_CONFIG_FILE_PATH: /etc/temporal/config/[Link]
ports:
- "7233:7233"
networks: [ml-ops-net]
deploy:
placement:
constraints: [[Link] == manager]

temporal-ui:
image: temporalio/ui:2.26
environment:
TEMPORAL_ADDRESS: temporal:7233
ports:
- "8088:8080"
networks: [ml-ops-net]

Human-in-the-Loop Annotation Workflow

# Temporal workflow: CVAT annotation → review → train
from temporalio import workflow, activity
from datetime import timedelta

@[Link]
class AnnotationToTrainingWorkflow:

@[Link]
async def run(self, dataset_id: str):
# 1. Submit to CVAT for annotation
task_id = await workflow.execute_activity(
 create_cvat_task, dataset_id, start_to_close_timeout=timedelta(minutes=5)
)

# 2. Wait for human annotation (could take days — Temporal handles this durably)
await workflow.execute_activity(
wait_for_annotation_complete, task_id,
start_to_close_timeout=timedelta(days=7),
heartbeat_timeout=timedelta(hours=1)
)

# 3. Human review signal (sent from your UI or Airflow)

await workflow.wait_condition(lambda: self._approved)

# 4. Trigger Prefect training flow

await workflow.execute_activity(
trigger_prefect_training_flow, dataset_id,
start_to_close_timeout=timedelta(hours=4)
)

@[Link]
def approve(self): self._approved = True
@[Link]
def reject(self): self._approved = False
 11 Phase 11 — Observability (Week 11)

11.1 — cAdvisor + NVIDIA DCGM Exporter

Docker Swarm Deployment
# [Link]
services:
cadvisor:
image: [Link]/cadvisor/cadvisor:v0.49.1
volumes:
- /:/rootfs:ro
- /var/run:/var/run:rw
- /sys:/sys:ro
- /var/lib/docker/:/var/lib/docker:ro
ports:
- "8081:8080"
networks: [ml-obs-net]
deploy:
mode: global # runs on every node

dcgm-exporter:
image: [Link]/nvidia/k8s/dcgm-exporter:3.3.5-3.4.0-ubuntu22.04
runtime: nvidia
environment:
NVIDIA_VISIBLE_DEVICES: all
ports:
- "9400:9400"
networks: [ml-obs-net]
deploy:
mode: global
placement:
constraints: [[Link] == true]

prometheus:
image: prom/prometheus:v2.53
volumes:
- ./[Link]:/etc/prometheus/[Link]
- prometheus_data:/prometheus
ports:
- "9090:9090"
networks: [ml-obs-net]
deploy:
placement:
constraints: [[Link] == manager]

grafana:
image: grafana/grafana:11.1.0
volumes:
- grafana_data:/var/lib/grafana
- ./grafana/provisioning:/etc/grafana/provisioning
ports:
- "3001:3000"
networks: [ml-obs-net]
deploy:
placement:
constraints: [[Link] == manager]

Key Grafana Dashboards to Import

NVIDIA GPU Overview (ID: GPU utilization, memory, temperature, power draw per node
12239)
 Docker Swarm & Containers Container CPU, memory, network, disk I/O
(cAdvisor)

Ray Dashboard integration Ray cluster utilization via custom Prometheus metrics

Airflow Metrics (ID: 11855) DAG run durations, task success rates, queue depth

Triton Server Metrics Inference throughput, latency, queue depth, GPU utilization per model

ClickHouse Overview Query rate, latency, storage usage, insert throughput

[Link] scrape targets

global:
scrape_interval: 15s

scrape_configs:
- job_name: cadvisor
static_configs: [{targets: ['cadvisor:8080']}]

- job_name: dcgm
static_configs: [{targets: ['node-2:9400', 'node-3:9400']}]

- job_name: triton
static_configs: [{targets: ['triton:8002']}]

- job_name: ray
static_configs: [{targets: ['ray-head:8080']}] # Ray exposes Prometheus metrics

- job_name: rabbitmq
static_configs: [{targets: ['rabbitmq:15692']}] # RabbitMQ Prometheus plugin

- job_name: airflow
static_configs: [{targets: ['airflow-webserver:8080']}]
 12 Phase 12 — Data Catalog (Week 12)

12.1 — OpenMetadata
Purpose
Central catalog for all data assets: Delta Lake tables, ClickHouse tables, ML models, pipelines, and feature views.
Provides data lineage (end-to-end: RabbitMQ message → Delta table → Feast feature → Triton model),
ownership, documentation, and discovery.

Docker Swarm Deployment

# [Link]
services:
openmetadata-server:
image: openmetadata/server:1.4.0
environment:
SERVER_PORT: 8585
SERVER_ADMIN_PORT: 8586
DB_HOST: postgres
DB_PORT: 5432
DB_USER: openmetadata_user
DB_USER_PASSWORD_SECRET: /run/secrets/om_pg_pass
DB_DATABASE: openmetadata
SEARCH_HOST: opensearch
SEARCH_PORT: 9200
ports:
- "8585:8585"
networks: [ml-ops-net]
deploy:
placement:
constraints: [[Link] == manager]

opensearch:
image: opensearchproject/opensearch:2.14
environment:
[Link]: single-node
DISABLE_SECURITY_PLUGIN: "true"
volumes:
- opensearch_data:/usr/share/opensearch/data
networks: [ml-ops-net]
deploy:
placement:
constraints: [[Link] == manager]

Connectors to Configure
Delta Lake / MinIO Use the S3/Datalake connector — discovers Parquet/Delta metadata automatically

ClickHouse Built-in ClickHouse connector — auto-profiles tables, column types, row counts

MLflow MlflowConnector — imports experiment runs, model versions, and tags

Apache Spark Spark lineage via OpenLineage integration in your Spark jobs

Airflow OpenMetadata Airflow plugin — DAG runs appear as pipeline lineage

Prefect HTTP connector via OpenMetadata REST API from Prefect flow hooks

OpenLineage Integration with Spark

# Add OpenLineage to Spark jobs for automatic lineage tracking
spark = [Link] \
.config("[Link]",
"[Link]") \
.config("[Link]", "http") \
.config("[Link]",
"[Link] \
.getOrCreate()
 13 Phase 13 — Data Visualization (Week 13)

13.1 — Metabase
Purpose
Self-hosted BI for non-technical stakeholders. Connect to ClickHouse for model monitoring dashboards,
PostgreSQL for operational metrics, and expose pre-built dashboards to business users.

Docker Swarm Deployment

# [Link]
services:
metabase:
image: metabase/metabase:v0.50.0
environment:
MB_DB_TYPE: postgres
MB_DB_DBNAME: metabase
MB_DB_PORT: 5432
MB_DB_USER: metabase_user
MB_DB_HOST: postgres
MB_DB_PASS_FILE: /run/secrets/metabase_pg_pass
ports:
- "3002:3000"
networks: [ml-data-net, ml-ops-net]
deploy:
placement:
constraints: [[Link] == manager]

ClickHouse Driver Setup

Metabase does not ship with ClickHouse by default. Add the community driver:
# Add to Metabase Docker image or volume mount
# Download from: [Link]
# Place in /plugins directory

# Or use the official ClickHouse Metabase image

image: clickhouse/metabase-driver:latest

Recommended Dashboards
Model Performance Overview Prediction accuracy, drift scores, latency — connects to ClickHouse
[Link]

Pipeline Health DAG success rates, processing volumes, SLA compliance — from ClickHouse
ops.pipeline_runs

Data Quality Report Missing values, outliers per feature over time — from Evidently via ClickHouse

GPU Utilization (via Copy GPU metrics from Prometheus into ClickHouse for SQL-queryable
ClickHouse) history
 14 Phase 14 — Vector DB (Week 14)

14.1 — Qdrant
Purpose
High-performance vector database for semantic search, similarity-based retrieval, embedding storage, and RAG
(Retrieval-Augmented Generation) pipelines.

Docker Swarm Deployment

# [Link]
services:
qdrant:
image: qdrant/qdrant:v1.10.0
volumes:
- qdrant_data:/qdrant/storage
- ./[Link]:/qdrant/config/[Link]
ports:
- "6333:6333" # HTTP REST
- "6334:6334" # gRPC
networks: [ml-serve-net, ml-train-net]
deploy:
placement:
constraints: [[Link] == true]

# [Link]
storage:
storage_path: /qdrant/storage
service:
max_request_size_mb: 64
max_workers: 0 # auto-detect
optimizers_config:
default_segment_number: 4

Usage Patterns
Embedding indexing (Prefect After training, extract embeddings from Triton, upsert into Qdrant collection
flow)

Similarity search BentoML service queries Qdrant before Triton for cache-hit or retrieval
augmentation

Feature-based ANN Store Feast features as vectors for nearest-neighbour lookup at serving time

RAG pipeline Store document chunks as vectors; query at inference time for context injection

# Index embeddings from model

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

client = QdrantClient(host="qdrant", port=6333)

client.create_collection(
collection_name="image_embeddings",
vectors_config=VectorParams(size=2048, distance=[Link])
)

# Upsert from Triton inference output

[Link](
collection_name="image_embeddings",
points=[PointStruct(id=item_id, vector=embedding, payload={"label": label})]
)
Complete Stack Reference

Layer Service Role / Notes

Storage Delta Lake on MinIO ACID lakehouse table format —

raw/bronze/silver/gold zones

Storage LakeFS Git branching for data — branch, merge, rollback data
loads

Ingestion RabbitMQ AMQP message broker — Airflow Celery broker + ML

event bus

Processing Apache Spark Batch ETL, Delta Lake transforms, medallion pipeline

Processing Ray Distributed ML training, GPU orchestration,

hyperparameter tuning

Database PostgreSQL Metadata store for all services — Airflow, Prefect,

Temporal, MLflow, Feast, OpenMetadata

Database ClickHouse OLAP analytics — model monitoring, pipeline metrics,

dashboards

Database Redis Online feature store (Feast), Celery broker, caching

Feature Store Feast Offline (Delta/MinIO) + online (Redis) feature store

Training JupyterHub Multi-user GPU notebooks — full stack access

Training MLflow (existing) Experiment tracking + model registry

Serving NVIDIA Triton GPU inference server — ONNX, TensorRT, PyTorch,

TF

Serving BentoML Model packaging + Python serving — wraps Triton

Monitoring Evidently AI Data drift + model monitoring — writes to ClickHouse

Orchestration Airflow (existing) Scheduled batch DAGs — Spark, dbt, GE, ETL

Orchestration Prefect (existing) ML training flows — Ray integration, dynamic

pipelines

Orchestration Temporal Durable long-running workflows — human-in-the-loop

(CVAT)

Observability cAdvisor Container-level CPU/memory/network metrics

Observability DCGM Exporter Per-container GPU metrics — utilization, memory,

temperature

Observability Prometheus + Grafana Metrics collection + dashboards

Security HashiCorp Vault Secrets management — all credentials fetched at

runtime

Catalog OpenMetadata + Data catalog, lineage, discovery, governance

OpenSearch

Visualization Metabase BI dashboards — connects to ClickHouse and

PostgreSQL

Vector DB Qdrant Embedding storage and similarity search

Annotation CVAT (existing) Computer vision annotation — feeds Temporal

workflows
Layer Service Role / Notes

Registry MinIO (existing) S3-compatible object storage — all data and model
artifacts
Rollout Timeline

Week Phase Deliverables

1 Foundation Vault, PostgreSQL, Redis — all future services depend on these

2 Lakehouse Delta Lake zones on MinIO, LakeFS, bucket structure

3 Streaming RabbitMQ, exchanges/queues, Airflow Celery integration

4 Processing Spark standalone cluster + Ray cluster across GPU nodes

5 Databases ClickHouse deployment, schema setup, Metabase pre-wiring

6 Feature Store Feast with Delta offline + Redis online store, materialization DAG

7 Training JupyterHub with GPU access, ML notebook image, Prefect-Ray flows

8 Serving Triton on GPU nodes, BentoML service, MLflow→ONNX→Triton pipeline

9 Monitoring Evidently monitoring flows, ClickHouse sink, retraining triggers

10 Orchestration Temporal, human-in-the-loop CVAT workflow, orchestrator wiring

11 Observability cAdvisor + DCGM on all nodes, Prometheus scraping, Grafana boards

12 Catalog OpenMetadata + OpenSearch, connector setup, OpenLineage in Spark

13 Visualization Metabase dashboards — model perf, pipeline health, data quality

14 Vector DB Qdrant, embedding indexing flow, BentoML retrieval integration

⚠ This timeline assumes one person per phase. With a team, phases 3-5 can run in parallel after Week 2, compressing
the timeline to ~8-9 weeks.

📝 Always deploy Vault and PostgreSQL in Week 1 regardless of team size. Every other service depends on them.
Rushing this foundation is the most common cause of painful rework later.

Important Operational Notes

Docker Swarm Constraints

Apply node labels to control service placement. Add GPU labels to worker nodes:
docker node update --label-add gpu=true node-2
docker node update --label-add gpu=true node-3
docker node update --label-add clickhouse=true node-3

Port Conflict Resolution

With this many services, port collisions are inevitable. Recommended remappings:
MinIO Remap to 9001:9000 (host:container) to free 9000 for ClickHouse native TCP

Grafana Use 3001:3000 — port 3000 may be used by other services

Metabase Use 3002:3000 — same reason

Triton gRPC Use 8003 — avoid conflict with Airflow (8080) and cAdvisor (8081)
GPU Swarm Gotcha
⚠ Docker Swarm does not natively support the NVIDIA Container Runtime in the same way Kubernetes does. You must
set the default runtime to nvidia on each GPU worker node in /etc/docker/[Link]: { "default-runtime": "nvidia",
"runtimes": { "nvidia": { "path": "nvidia-container-runtime" } } } — then services using 'runtime: nvidia' will work correctly in
Swarm stacks.

Kubernetes Migration Path

Once this stack is stable, consider migrating to K3s or RKE2. Ray, Triton, and Spark have far better Kubernetes
operator support. The migration path is: Docker Swarm stacks → Helm charts — your Vault paths, MinIO buckets,
and PostgreSQL databases carry over unchanged.