Training

This guide covers the complete model training workflow using the ViewAI Python SDK.

Overview

The ViewAI SDK enables you to train machine learning models on the ViewAI platform using your own datasets. The platform handles:

• Automated feature engineering

• Model selection and hyperparameter tuning

• Training infrastructure and scaling

• Model versioning and tracking

Quick Start

from viewai_client import ViewAIClient
import pandas as pd

# Initialize client
client = ViewAIClient(api_key="your-api-key")

# Prepare data
df = pd.read_csv("training_data.csv")

# Initiate training
job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    model_name="Customer Churn Model",
    wait_for_completion=True
)

print(f"Training complete! Model ID: {job.job_id}")

Training Workflow

The typical training workflow consists of these steps:

graph LR
    A[Prepare Data] --> B[Set Workspace/Project]
    B --> C[Initiate Training]
    C --> D[Monitor Progress]
    D --> E[Training Complete]
    E --> F[Deploy Model]

Complete example (step-by-step):

1

Initialize client

from viewai_client import ViewAIClient
import pandas as pd

# Initialize client
client = ViewAIClient(api_key="your-api-key")

2

Set workspace context

workspace = client.retrieve_workspace_by_name("production")
client.set_current_workspace(workspace)

3

Prepare training data

df = pd.read_csv("customer_data.csv")

4

Initiate training

job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace=workspace,
    model_name="Customer Churn Model",
    description="Trained on Q4 2024 data",
    wait_for_completion=True
)

5

Check training results

if job.status == "training_completed":
    print(f"Success! View at: {job.dashboard_url}")

Preparing Data

Data Requirements

Your training dataset must meet these requirements:

• Format: Pandas DataFrame with labeled columns

• Target column: Must be present in the DataFrame

• Size: At least 100 rows recommended for meaningful training

• Missing values: Handle or remove before training

Basic Data Preparation

import pandas as pd
import numpy as np

# Load data
df = pd.read_csv("customer_data.csv")

# Check data shape
print(f"Dataset shape: {df.shape}")
print(f"Columns: {list(df.columns)}")

# Check for missing values
print(f"Missing values:\n{df.isnull().sum()}")

# Handle missing values
df = df.dropna()  # Or use df.fillna()

# Verify target column exists
target_column = "churn"
if target_column not in df.columns:
    raise ValueError(f"Target column '{target_column}' not found")

print(f"Target distribution:\n{df[target_column].value_counts()}")

Advanced Data Preparation

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split

# Load and explore data
df = pd.read_csv("customer_data.csv")

# Remove unnecessary columns
df = df.drop(["customer_id", "signup_date"], axis=1)

# Handle missing values

# For numeric columns: fill with median
numeric_cols = df.select_dtypes(include=[np.number]).columns
df[numeric_cols] = df[numeric_cols].fillna(df[numeric_cols].median())

# For categorical columns: fill with mode
categorical_cols = df.select_dtypes(include=["object"]).columns
for col in categorical_cols:
    df[col] = df[col].fillna(df[col].mode()[0])

# Encode target variable if needed
if df["churn"].dtype == "object":
    df["churn"] = df["churn"].map({"Yes": 1, "No": 0})

# Check class balance
print(f"Class distribution:\n{df['churn'].value_counts(normalize=True)}")

# Handle class imbalance if needed
if df["churn"].value_counts(normalize=True).min() < 0.1:
    print("Warning: Significant class imbalance detected")
    # Consider resampling or adjusting class weights

# Verify data is ready
print(f"Final dataset shape: {df.shape}")
print(f"Missing values: {df.isnull().sum().sum()}")

Creating Synthetic Training Data

For testing or demonstration:

import pandas as pd
import numpy as np

def create_sample_dataset(n_samples=1000):
    """Create synthetic customer churn dataset."""
    np.random.seed(42)

    data = {
        "age": np.random.randint(18, 80, n_samples),
        "tenure_months": np.random.randint(1, 60, n_samples),
        "monthly_charges": np.random.uniform(20, 120, n_samples),
        "total_charges": np.random.uniform(100, 5000, n_samples),
        "contract_type": np.random.choice(
            ["Month-to-month", "One year", "Two year"],
            n_samples
        ),
        "internet_service": np.random.choice(
            ["DSL", "Fiber optic", "No"],
            n_samples
        ),
        "payment_method": np.random.choice(
            ["Electronic check", "Mailed check", "Bank transfer"],
            n_samples
        )
    }

    df = pd.DataFrame(data)

    # Create target based on features
    churn_prob = (
        (df["contract_type"] == "Month-to-month") * 0.3 +
        (df["tenure_months"] < 12) * 0.2 +
        (df["monthly_charges"] > 80) * 0.1 +
        np.random.normal(0, 0.1, n_samples)
    )
    df["churn"] = (churn_prob > 0.4).astype(int)

    return df

# Create dataset
df = create_sample_dataset(1000)
print(f"Created dataset with {len(df)} samples")

Initiating Training

Basic Training Job

# Initialize client with workspace
client = ViewAIClient(api_key="your-api-key")
workspace = client.retrieve_default_workspace()

# Initiate training
job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace=workspace,
    model_name="Customer Churn Model",
    description="Baseline churn prediction model",
    wait_for_completion=True
)

print(f"Training job ID: {job.job_id}")
print(f"Status: {job.status}")

Training with Project Context

Organize models by project:

# Get workspace and project
workspace = client.retrieve_workspace_by_name("production")
project = client.retrieve_project_by_name("customer-analytics")

# Initiate training in project context
job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace=workspace,
    project=project,
    model_name="Churn Model v2.0",
    description="Updated model with new features",
    wait_for_completion=True
)

Using the Training Service Directly

For more control, use the training service:

# Access training service
training_service = client.training_service

# Initiate training
job = training_service.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace="workspace-id-123",  # Can use ID string
    model_name="My Model",
    description="Model description",
    wait_for_completion=False
)

# Monitor separately
training_service.monitor_training_job_until_complete(job)

Asynchronous Training

For large datasets, start training without waiting:

# Submit training job
job = client.initiate_training_job(
    dataset=large_df,
    target_column="target",
    workspace=workspace,
    model_name="Large Dataset Model",
    wait_for_completion=False  # Don't wait
)

print(f"Training started: {job.job_id}")
print(f"Monitor at: {job.dashboard_url}")

# Check status later
status = client.training_service.retrieve_training_job_status(job.job_id)
print(f"Current status: {status.get('status')}")

Monitoring Progress

Automatic Monitoring

When wait_for_completion=True, training progress is displayed automatically:

job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace=workspace,
    model_name="My Model",
    wait_for_completion=True  # Shows progress bar
)

# Output:

# Training Progress: 100%|██████████| 30/30 [02:15<00:00]

# ✓ Training completed! View model at: https://app.viewai.ca/dashboard/...

Manual Monitoring

Monitor training progress manually:

# Start training without waiting
job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace=workspace,
    model_name="My Model",
    wait_for_completion=False
)

# Monitor with custom settings
completed_job = client.training_service.monitor_training_job_until_complete(
    job=job,
    poll_interval=5,      # Check every 5 seconds
    max_wait_time=600     # Wait up to 10 minutes
)

print(f"Final status: {completed_job.status}")

Checking Training Status

Retrieve training job status at any time:

# Get status by job ID
status = client.training_service.retrieve_training_job_status(
    job_id="job-123abc"
)

print(f"Status: {status.get('status')}")
print(f"Progress: {status.get('progress', 'N/A')}")

Possible status values:

• "pending": Job submitted, not yet started

• "training": Model training in progress

• "training_completed": Training finished successfully

• "failed": Training failed (check error message)

Retrieving Training Results

Get detailed results from completed training:

# Get training results
results = client.training_service.retrieve_training_job_results(
    job_id=job.job_id
)

if results:
    print(f"Model ID: {results.get('model_id')}")
    print(f"Status: {results.get('status')}")

    # Get metrics if available
    metrics = results.get('metrics', {})
    if metrics:
        print(f"Accuracy: {metrics.get('accuracy')}")
        print(f"Precision: {metrics.get('precision')}")
        print(f"Recall: {metrics.get('recall')}")

Training Job Management

The TrainingJob Class

The TrainingJob class represents a training job:

# TrainingJob attributes
print(f"Job ID: {job.job_id}")           # Unique job identifier
print(f"Model Name: {job.model_name}")   # Model name
print(f"Status: {job.status}")           # Current status
print(f"Workspace: {job.workspace_id}")  # Workspace ID
print(f"Dashboard: {job.dashboard_url}") # Dashboard URL

# String representation
print(job)

# Output: TrainingJob(job_id='job-123', model_name='My Model', status='training_completed')

Managing Multiple Training Jobs

Track multiple training jobs:

# Start multiple training jobs
jobs = []

for model_config in model_configs:
    job = client.initiate_training_job(
        dataset=model_config["data"],
        target_column=model_config["target"],
        workspace=workspace,
        model_name=model_config["name"],
        wait_for_completion=False
    )
    jobs.append(job)
    print(f"Started: {job.model_name} ({job.job_id})")

# Monitor all jobs
for job in jobs:
    status = client.training_service.retrieve_training_job_status(
        job.job_id
    )
    print(f"{job.model_name}: {status.get('status')}")

Canceling Training Jobs

Cancel a running training job (if supported):

# Cancel training job
success = client.training_service.cancel_training_job(
    job_id="job-123abc"
)

if success:
    print("Training job cancelled")
else:
    print("Failed to cancel job or not supported")

Listing Training Jobs

List all training jobs in a workspace:

# List training jobs for workspace
jobs = client.training_service.list_training_jobs_for_workspace(
    workspace_id="ws-123"
)

for job_info in jobs:
    print(f"{job_info.get('name')}: {job_info.get('status')}")

Error Handling

Common Training Errors

Handle training errors gracefully:

from viewai_client.exceptions import (
    ViewAIError,
    ValidationError,
    APIError
)

try:
    job = client.initiate_training_job(
        dataset=df,
        target_column="churn",
        workspace=workspace,
        model_name="My Model",
        wait_for_completion=True
    )

except ValidationError as e:
    print(f"Data validation failed: {e.message}")
    print(f"Details: {e.details}")
    # Fix data and retry

except APIError as e:
    print(f"API error: {e.message}")
    print(f"Status code: {e.status_code}")
    # Check API status and retry

except ViewAIError as e:
    print(f"Training error: {e.message}")

except Exception as e:
    print(f"Unexpected error: {e}")

Validating Input Data

Validate data before training:

def validate_training_data(df, target_column):
    """Validate training data before submission."""
    errors = []

    # Check DataFrame
    if df is None or df.empty:
        errors.append("DataFrame is empty")

    # Check target column
    if target_column not in df.columns:
        errors.append(f"Target column '{target_column}' not found")

    # Check for sufficient data
    if len(df) < 100:
        errors.append("Dataset too small (minimum 100 rows)")

    # Check for missing values
    if df.isnull().sum().sum() > 0:
        errors.append("Dataset contains missing values")

    # Check class balance for classification
    if target_column in df.columns:
        class_dist = df[target_column].value_counts(normalize=True)
        if class_dist.min() < 0.01:
            errors.append("Severe class imbalance detected")

    return errors

# Validate before training
errors = validate_training_data(df, "churn")
if errors:
    print("Validation errors:")
    for error in errors:
        print(f"  - {error}")
else:
    job = client.initiate_training_job(
        dataset=df,
        target_column="churn",
        workspace=workspace,
        model_name="My Model"
    )

Handling Training Failures

Handle failed training jobs:

# Initiate training
job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace=workspace,
    model_name="My Model",
    wait_for_completion=True
)

# Check final status
if job.status == "training_completed":
    print("Training successful!")
    print(f"Model URL: {job.dashboard_url}")

elif job.status == "failed":
    print("Training failed!")
    print(f"Check details: {job.dashboard_url}")

    # Get detailed error information
    results = client.training_service.retrieve_training_job_results(
        job.job_id
    )
    if results:
        error_msg = results.get("error_message", "Unknown error")
        print(f"Error: {error_msg}")

else:
    print(f"Training in unexpected state: {job.status}")

Retry Logic for Training

Implement retry logic for transient failures:

from time import sleep

def train_with_retry(client, df, target_column, workspace, model_name, max_retries=3):
    """Train model with retry logic."""
    for attempt in range(max_retries):
        try:
            print(f"Training attempt {attempt + 1}/{max_retries}")

            job = client.initiate_training_job(
                dataset=df,
                target_column=target_column,
                workspace=workspace,
                model_name=f"{model_name} (attempt {attempt + 1})",
                wait_for_completion=True
            )

            if job.status == "training_completed":
                return job

            print(f"Training failed: {job.status}")

        except Exception as e:
            print(f"Error on attempt {attempt + 1}: {e}")

        if attempt < max_retries - 1:
            wait_time = 2 ** attempt  # Exponential backoff
            print(f"Retrying in {wait_time} seconds...")
            sleep(wait_time)

    raise RuntimeError("Training failed after all retries")

Best Practices

Best practices (each step contains guidance and examples):

1

Validate data before training

Always validate your data.

# Validation checklist
def pre_training_validation(df, target_column):
    """Comprehensive pre-training validation."""
    checks = {
        "non_empty": len(df) > 0,
        "sufficient_size": len(df) >= 100,
        "target_exists": target_column in df.columns,
        "no_missing_target": df[target_column].isnull().sum() == 0,
        "no_duplicates": df.duplicated().sum() == 0,
        "valid_types": all(df.dtypes != 'object' or df[col].nunique() < 1000
                          for col in df.columns)
    }

    print("Pre-training validation:")
    for check, passed in checks.items():
        status = "✓" if passed else "✗"
        print(f"  {status} {check}")

    return all(checks.values())

# Run validation
if pre_training_validation(df, "churn"):
    job = client.initiate_training_job(dataset=df, ...)

2

Use meaningful model names

Use descriptive, versioned names.

from datetime import datetime

# Generate descriptive model name
model_name = f"Churn_v{version}_{datetime.now().strftime('%Y%m%d')}"
description = f"Customer churn model v{version} - trained on {data_date}"

job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace=workspace,
    model_name=model_name,
    description=description
)

3

Set workspace context early

Set workspace context at the start.

# Set workspace context once
client = ViewAIClient(api_key="your-api-key")
workspace = client.retrieve_workspace_by_name("production")
client.set_current_workspace(workspace)

# Now training jobs use this workspace by default
job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    model_name="Model 1"
    # workspace parameter not needed
)

4

Monitor async training jobs

For async training, implement proper monitoring.

# Start training
job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace=workspace,
    model_name="My Model",
    wait_for_completion=False
)

# Store job ID for later reference
job_id = job.job_id
print(f"Training job started: {job_id}")

# Later: check status
status = client.training_service.retrieve_training_job_status(job_id)

if status.get('status') == 'training_completed':
    print("Training complete!")
elif status.get('status') == 'failed':
    print("Training failed - check dashboard")
else:
    print(f"Training in progress: {status.get('status')}")

5

Handle large datasets

For large datasets, optimize data transfer.

import pandas as pd

# Load and prepare data efficiently
df = pd.read_csv("large_dataset.csv", low_memory=False)

# Select only necessary columns
feature_columns = ["age", "income", "tenure", "churn"]
df = df[feature_columns]

# Handle missing values before upload
df = df.dropna()

# Start async training for large datasets
job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace=workspace,
    model_name="Large Dataset Model",
    wait_for_completion=False  # Don't wait for large jobs
)

print(f"Job submitted: {job.job_id}")

6

Track training experiments

Keep track of training experiments.

import json
from datetime import datetime

# Training experiment log
experiment_log = {
    "timestamp": datetime.now().isoformat(),
    "model_name": "Churn Model v1.0",
    "dataset_size": len(df),
    "target_column": "churn",
    "features": list(df.columns),
    "class_distribution": df["churn"].value_counts().to_dict()
}

# Start training
job = client.initiate_training_job(
    dataset=df,
    target_column="churn",
    workspace=workspace,
    model_name=experiment_log["model_name"]
)

# Add job info to log
experiment_log["job_id"] = job.job_id
experiment_log["status"] = job.status

# Save experiment log
with open(f"experiments/{job.job_id}.json", "w") as f:
    json.dump(experiment_log, f, indent=2)

7

Test with small datasets first

Test your training pipeline with a small dataset.

# Test with small sample
test_df = df.sample(n=min(100, len(df)), random_state=42)

job = client.initiate_training_job(
    dataset=test_df,
    target_column="churn",
    workspace=workspace,
    model_name="Test Model (small sample)",
    wait_for_completion=True
)

if job.status == "training_completed":
    print("Test successful - proceed with full dataset")

    # Train on full dataset
    full_job = client.initiate_training_job(
        dataset=df,
        target_column="churn",
        workspace=workspace,
        model_name="Production Model"
    )

See Also

• Model Deployment - Deploy trained models

• Workspace Management - Organize your models

• Error Handling - Comprehensive error handling

• API Reference - Detailed API documentation