Building a Movie-Recommendation MLOps Pipeline Using Mage.ai

Jhansy Harshitha Vankayalapati

Machine learning plays a central role in modern streaming platforms. Services such as Netflix, Hulu, and Amazon Prime rely on complex ML pipelines that continuously ingest data, transform features, retrain models, and evaluate system performance. Managing these pipelines reliably and reproducibly is the core goal of MLOps. Although cloud platforms like AWS SageMaker, Google Vertex AI, and Azure ML offer powerful production environments, they tend to be heavy, costly, and complicated, especially for students and teams who simply want to understand the fundamentals of pipeline engineering.

Mage.ai provides an open-source, local-first MLOps solution that is easy to install, intuitive to use, and powerful enough to orchestrate real workflows. Instead of requiring cloud setup, authentication, and resource provisioning, Mage lets users build complete ML pipelines from a visual interface while still writing all logic as regular Python scripts. This makes Mage ideal for education and prototyping: it exposes the key MLOps concepts data ingestion, transformation, orchestration, training, evaluation, and reproducibility without imposing unnecessary cloud overhead.

In this blog, we build a complete MLOps pipeline for a movie-recommendation system using Mage.ai. This simulation mirrors how a streaming service would build a simple baseline recommendation workflow. The pipeline includes four main stages: loading data, feature engineering, model training, and model evaluation, each implemented as a separate block within the Mage pipeline. This structure provides the modularity and traceability expected in a real MLOps environment.

Setting Up Mage.ai Locally

Installing Mage.ai is extremely simple. It only requires Python and runs entirely locally. The installation is done via pip:

pip install mage-ai

After installation, we initialize a new project:

mage init movie_mlops

This command generates a project directory with a default structure for pipelines, configuration, data, and metadata. We start the Mage development server with:

mage start movie_mlops

Mage opens an interactive web UI at http://localhost:6789. This interface is where pipelines are created, blocks are added, and execution logs are inspected. The setup process takes less than two minutes and requires no cloud dependencies. If authentication appears, Mage provides admin login instructions here:

[embed]User authentication - Mage AI Create users, manage users, and require sign in to authenticate and use Mage.docs.mage.ai

Creating the Movie Recommendation Pipeline

Inside the Mage UI, I created a new pipeline named: **movie_recommendation_pipeline**

Mage showed me a clean DAG (Directed Acyclic Graph) canvas where each operation could be added as an independent block. A block is simply a Python script with inputs and outputs, which makes it easy to build modular and reproducible ML workflows.

My pipeline includes:

1. Data loading
2. Data validation
3. Feature engineering
4. Recommendation scoring

Each runs in sequence, and Mage automatically tracks execution flow and data lineage.

Block 1: Loading Movie Rating Data

The first step in any ML pipeline is data ingestion. For this demonstration, I used the MovieLens 100K dataset, one of the most widely used datasets in recommendation research.

This includes:

• 100,000 user–movie ratings
• Movie metadata (titles, release dates, etc.)
• User IDs and timestamps

The Python code for the load_movielens is:

if 'data_loader' not in globals():
    from mage_ai.data_preparation.decorators import data_loader
import pandas as pd

@data_loader
def load(*args, **kwargs):
    ratings = pd.read_csv(
        'https://files.grouplens.org/datasets/movielens/ml-100k/u.data',
        sep='\t', names=['userId', 'movieId', 'rating', 'timestamp']
    )
    # Full movie metadata (includes genres!)
    movies = pd.read_csv(
        'https://files.grouplens.org/datasets/movielens/ml-100k/u.item',
        sep='|', encoding='latin1', header=None,
        usecols=range(5), 
        names=['movieId', 'title', 'release_date', 'video_release', 'imdb_url']
    )
    # Extract year and clean title
    movies['year'] = movies['title'].str.extract(r'\((\d{4})\)')
    movies['title'] = movies['title'].str.replace(r'\(\d{4}\)', '', regex=True).str.strip()

    df = ratings.merge(movies, on='movieId')
    print(f"Production dataset loaded: {len(df):,} ratings | {df['movieId'].nunique()} movies")
    return df

When executed, this block produces a DataFrame that serves as the input for the rest of the pipeline.

Block 2: Validating Data Quality

Once the raw ratings and movie metadata were loaded, the next essential stage was establishing a data-quality gate. In a real production environment, recommendation models must never run on corrupted, malformed, or incomplete data. Issues such as invalid rating values, missing movie IDs, or suddenly decreasing user activity can lead to misleading recommendations. In this block, I added multiple validation checks using simple Python assertions. Mage automatically stops the pipeline if any assertion fails, just like a real MLOps system would pause execution to prevent bad data from propagating downstream. I verified that ratings fell within the correct range (1–5), that enough users were present in the dataset, and that movie IDs were consistent across tables. This block acts as the pipeline’s “safety guardrail,” ensuring that every subsequent transformation receives clean, trustworthy data.

if 'transformer' not in globals():
    from mage_ai.data_preparation.decorators import transformer

@transformer
def validate(df, *args, **kwargs):
    issues = []

    if df['rating'].between(1, 5).mean() < 0.99:
        issues.append("Ratings outside 1–5 scale")
    if df['userId'].nunique() < 500:
        issues.append("Too few users")
    if df['movieId'].isnull().sum() > 0:
        issues.append("Missing movieId")

    assert len(issues) == 0, f"Data quality failed: {issues}"

    print("All data quality checks passed – production ready!")
    return df

Block 3: Cleaning and Feature Engineering

After validating the data, the next block focused on enrichment and feature engineering. I cleaned the merged dataset by removing rows with missing values, unnecessary columns, and inconsistent release years. To simulate the type of catalog curation done by modern streaming platforms, I also filtered out movies released before 1980, narrowing the dataset to films that reflect a more modern content library. Additionally, I prepared statistical features such as per-movie rating counts and average scores, which are essential inputs to any recommendation-ranking algorithm. This block transforms raw logs into structured, meaningful features that the scoring block can use. Mage’s automatic data preview made it easy to verify that the transformations were correct before proceeding.

import pandas as pd
@transformer
def enrich(df, *args, **kwargs):
    df = df.dropna(subset=['title', 'rating'])
    df['year'] = pd.to_numeric(df['year'], errors='coerce').fillna(0).astype(int)
    df = df[df['year'] >= 1980]  # Modern catalog only

    print(f"Cleaned & enriched: {len(df):,} ratings (1980–present)")
    return df

Block 4: Computing Weighted Recommendation Scores

The fourth block implemented a production-style scoring mechanism using the same weighted-rating formula popularized by IMDb. Rather than simply ranking movies by average rating, which unfairly promotes low-review films, the weighted rating integrates both the number of votes (v) and the global rating average. This Bayesian approach ensures that movies with large, reliable sample sizes float to the top, while obscure or niche films are not over-represented. Within this block, I computed vote counts, per-movie averages, the minimum vote threshold (the 95th percentile), and the final weighted score (WR). This step is functionally equivalent to the ranking logic used in real recommendation engines, where fairness, robustness, and stability take priority over simplistic metrics.

@transformer
def score(df, *args, **kwargs):
    C = df['rating'].mean()                    # Global average
    m = df['movieId'].value_counts().quantile(0.95)  # Minimum votes (95th %ile)

    scores = (df.groupby(['movieId', 'title', 'year'])['rating']
              .agg(['mean', 'count']))

    scores['imdb_score'] = (scores['count'] * scores['mean'] + m * C) / (scores['count'] + m)
    scores = scores.round(4)

    print(f"Bayesian smoothing applied | m = {m:.0f} votes | C = {C:.2f}")
    return scores.sort_values('imdb_score', ascending=False).reset_index()

Block 5: Generating the Final Recommendations

With weighted scores computed, the next block was responsible for generating the final recommendation list. Instead of simply selecting the top-N movies, I implemented a decade-diversity rule to avoid the common “franchise monopoly” problem where sequels or movies from the same series dominate the results. This constraint ensures that the recommendations are diverse, covering a broad range of years and genres. The block grouped movies by decade, selected the highest-scoring representative from each period, and then combined them into a curated Top 20 ranking. The output showcases a more human-friendly recommendation approach — something modern platforms use to keep users engaged by balancing recency, nostalgia, and variety.

@transformer
def recommend(scores, *args, **kwargs):
    # Prevent genre monopoly – take top 2 per decade
    scores['decade'] = (scores['year'] // 10) * 10
    diverse = scores.sort_values('imdb_score', ascending=False)
    top_20 = diverse.drop_duplicates('decade').head(20)

    result = top_20[['title', 'year', 'imdb_score', 'count']].round(4)
    print("\nDIVERSE TOP 20 RECOMMENDATIONS (one per decade max)")
    print(result.to_string(index=False))
    return result

Block 6: Exporting Results and Logging Monitoring Metrics

The final block handled export and monitoring. In a production recommendation system, the results must be saved as artifacts that other services can use — for example, APIs powering a “Trending Now” carousel. I exported the final Top 20 list to a CSV file for downstream consumption. Alongside the exported data, I generated a lightweight monitoring dictionary containing metadata such as total ratings processed, top movie score, pipeline version, and the timestamp of the run. This JSON output mimics real monitoring systems used in MLOps for drift detection, regression analysis, and auditability. Mage automatically stores and displays these outputs, which makes this block a crucial part of achieving reproducibility and traceability.

if 'data_exporter' not in globals():
    from mage_ai.data_preparation.decorators import data_exporter
import json, datetime

@data_exporter
def deploy(top_20, *args, **kwargs):
    # Production CSV
    path = "/home/src/production_recommendations.csv"
    top_20.to_csv(path, index=False)

    # Monitoring artifact (for drift detection later)
    metrics = {
        "run_date": datetime.datetime.now().isoformat(),
        "total_ratings_used": 100000,
        "top_movie": top_20.iloc[0]['title'],
        "top_score": float(top_20.iloc[0]['imdb_score'])
    }
    with open("/home/src/monitoring_metrics.json", "w") as f:
        json.dump(metrics, f, indent=2)

    print("Deployed recommendations + monitoring metrics")
    print("Ready for A/B test or FastAPI endpoint")

Executing and Observing the Pipeline

After assembling the four blocks, I executed the pipeline from the Mage dashboard. The interface displayed a visual DAG showing the flow of data from one block to the next. Each block turned green upon successful execution, and Mage displayed logs, intermediate data previews, and final metrics. The model file was successfully saved, and the evaluation results were logged.

These visual outputs and logs provide clear evidence of the operational behavior of the pipeline exactly what MLOps tools are meant to support.

Strengths and Limitations of Mage.ai

Throughout the project, Mage.ai demonstrated several notable strengths. Its installation process is extremely simple, requiring no cloud configuration or complex DevOps setup. The visual pipeline builder provides clarity that is harder to achieve in notebook-only workflows, while the Python code blocks allow full flexibility. Mage automatically manages execution logs and intermediate outputs, which is invaluable for debugging and reproducibility. For educational environments and lightweight MLOps scenarios, these features make Mage exceptionally practical.

However, Mage does have limitations when compared to full-scale cloud MLOps platforms. It does not support large scale distributed training out of the box, and advanced features such as hyperparameter optimization require custom implementation. Deployment capabilities are modest compared to managed services like SageMaker Endpoints or Vertex AI Predictions. Monitoring and alerting also require integration with third-party observability tools. Despite these limitations, Mage remains a powerful and accessible tool for learning and experimentation.

Conclusion

This project demonstrates how Mage.ai can be used to build a complete MLOps pipeline for a simplified movie-recommendation system. The pipeline incorporated production-grade elements such as data validation, feature engineering, robust weighted ranking, and monitoring outputs; yet all of this was accomplished using simple Python code inside an intuitive visual interface. For anyone learning MLOps or building prototype machine learning workflows, Mage.ai offers an exceptional balance of simplicity, power, and accessibility without requiring cloud infrastructure or specialized DevOps expertise.