Truecaller: MLOps Excellence & Cost Optimization at Billion-Row Scale

Truecaller: MLOps Excellence & Cost Optimization at Billion-Row Scale

Role: Senior Data Scientist

Duration: Apr 2022 - Present

---

Executive Summary

At Truecaller, I lead fraud detection and spam classification efforts for one of the world’s largest caller identification platforms. Working with datasets containing billions of rows, I architected MLOps pipelines that reduced cloud infrastructure costs by 25% while improving fraud detection rates by 8% and SMS spam detection by 10%. The key innovations: automated model retraining triggered by data drift, PySpark optimization, and semi-supervised learning for label-scarce domains.

---

The Problem: Scale, Cost, and Continuous Evolution

Business Context

Truecaller processes:

• 250+ million active users globally
• Billions of phone calls and SMS messages daily
• Real-time fraud detection for financial transactions
• Spam classification across multiple languages and regions

Technical Challenges

1. Data Scale

• Datasets with 5-10 billion rows per month
• 500+ features per transaction/message
• High-velocity data streams (1M+ events per minute peak)
• Storage costs growing 30% year-over-year

2. Model Staleness

• Fraud patterns evolve rapidly (new scam techniques weekly)
• Spam content changes with current events and trends
• Manual retraining was reactive, not proactive
• Model performance degraded 5-10% within 2-3 months

3. Cloud Cost Explosion

• Spark clusters running 24/7 for batch processing
• Inefficient queries scanning entire datasets
• Redundant transformations across multiple pipelines
• Monthly GCP costs increasing unsustainably

4. Label Scarcity

• SMS spam: only 2% of data labeled by users
• Cold-start problem for new spam categories
• Manual labeling doesn’t scale
• Need to leverage unlabeled data

---

The Solution: MLOps Automation + Spark Optimization

1. Automated Retraining Pipeline with Data Drift Detection

The Core Idea: Models should retrain automatically when data distribution shifts, not on a fixed schedule.

Architecture:

[Production Data Stream]
         ↓
[BigQuery - Data Warehouse]
         ↓
[Cloud Scheduler - Daily Drift Check]
         ↓
[Drift Detection Service]
    ├── Feature Distribution Analysis (KL Divergence)
    ├── Prediction Distribution Shift
    └── Performance Metric Degradation
         ↓
    [Drift Detected?] ──No──> [Continue Monitoring]
         ↓ Yes
    [Pub/Sub Topic - Retrain Trigger]
         ↓
[Cloud Function - Orchestrator]
         ↓
[MLFlow Pipeline]
    ├── Data Validation (Great Expectations)
    ├── Feature Engineering (PySpark)
    ├── Model Training (XGBoost/LightGBM)
    ├── Hyperparameter Tuning (Optuna)
    ├── Model Evaluation (Hold-out + Temporal Split)
    └── Model Registry (MLFlow)
         ↓
    [A/B Test Deployment]
         ↓
    [Production Serving (if metrics improve)]

Drift Detection Implementation:

import numpy as np
from scipy.stats import ks_2samp, entropy

class DataDriftDetector:
    def __init__(self, reference_data, threshold=0.05):
        self.reference_data = reference_data
        self.threshold = threshold

    def detect_feature_drift(self, current_data, feature_name):
        """
        Detect drift using Kolmogorov-Smirnov test for numerical features
        and KL divergence for categorical features
        """
        ref_values = self.reference_data[feature_name]
        curr_values = current_data[feature_name]

        if self._is_numerical(feature_name):
            # KS test for continuous distributions
            statistic, p_value = ks_2samp(ref_values, curr_values)
            drift_detected = p_value < self.threshold
        else:
            # KL divergence for categorical distributions
            ref_dist = self._get_distribution(ref_values)
            curr_dist = self._get_distribution(curr_values)
            kl_div = entropy(curr_dist, ref_dist)
            drift_detected = kl_div > 0.1  # Threshold tuned empirically

        return drift_detected

    def detect_prediction_drift(self, model, current_data):
        """
        Monitor if prediction distributions shift significantly
        """
        predictions = model.predict_proba(current_data)
        ref_predictions = self.reference_data['predictions']

        # Compare prediction score distributions
        statistic, p_value = ks_2samp(predictions[:, 1], ref_predictions)
        return p_value < self.threshold

    def detect_performance_drift(self, current_metrics, reference_metrics):
        """
        Alert if model performance degrades beyond acceptable threshold
        """
        metrics_to_monitor = ['auc', 'precision', 'recall', 'f1']
        for metric in metrics_to_monitor:
            degradation = (reference_metrics[metric] - current_metrics[metric]) / reference_metrics[metric]
            if degradation > 0.05:  # 5% degradation threshold
                return True
        return False

MLFlow Integration:

import mlflow
from mlflow.tracking import MlflowClient

class FraudModelPipeline:
    def __init__(self, experiment_name="fraud-detection"):
        mlflow.set_experiment(experiment_name)
        self.client = MlflowClient()

    def train_and_log_model(self, train_data, val_data, params):
        with mlflow.start_run():
            # Log parameters
            mlflow.log_params(params)

            # Train model
            model = self._train_model(train_data, params)

            # Evaluate
            metrics = self._evaluate_model(model, val_data)
            mlflow.log_metrics(metrics)

            # Log model artifact
            mlflow.sklearn.log_model(model, "model")

            # Log feature importance
            self._log_feature_importance(model)

            # Register model if performance improves
            if metrics['auc'] > self.get_production_auc():
                model_uri = f"runs:/{mlflow.active_run().info.run_id}/model"
                mlflow.register_model(model_uri, "fraud-detection-model")

    def get_production_auc(self):
        # Fetch current production model metrics
        production_model = self.client.get_latest_versions("fraud-detection-model", stages=["Production"])
        if production_model:
            run_id = production_model[0].run_id
            run = self.client.get_run(run_id)
            return run.data.metrics.get('auc', 0.0)
        return 0.0

Impact:

• Reduced time-to-deployment for model updates from 2 weeks to 8 hours
• Models now retrain automatically when drift exceeds thresholds
• Improved fraud detection rate by 8% through timely updates
• Eliminated manual monitoring overhead

---

2. PySpark Pipeline Optimization: 25% Cost Reduction

Problem: Existing Spark jobs were inefficient, leading to:

• Long-running clusters (8+ hours for daily batch jobs)
• Full table scans on multi-TB datasets
• Redundant shuffles and data movement
• High memory pressure causing OOM errors

Optimization Strategies:

a) Partition Pruning & Predicate Pushdown

# BEFORE: Inefficient query
fraud_data = spark.read.parquet("gs://bucket/fraud_events/")
filtered_data = fraud_data.filter(col("event_date") == "2024-01-15")

# AFTER: Partition pruning
fraud_data = spark.read.parquet("gs://bucket/fraud_events/") \
    .where("event_date = '2024-01-15'")  # Pushed to storage layer

# Result: 90% reduction in data scanned

b) Broadcast Joins for Small Dimension Tables

# BEFORE: Shuffle join (expensive for large fact table)
enriched = transactions.join(user_features, "user_id")

# AFTER: Broadcast join
from pyspark.sql.functions import broadcast
enriched = transactions.join(broadcast(user_features), "user_id")

# Result: Eliminated shuffle for 500M row table

c) Caching Strategic Intermediate Results

# BEFORE: Recomputing transformations multiple times
base_features = compute_base_features(raw_data)
agg_features_1 = base_features.groupBy("user_id").agg(...)
agg_features_2 = base_features.groupBy("session_id").agg(...)

# AFTER: Cache intermediate result
base_features = compute_base_features(raw_data).cache()
agg_features_1 = base_features.groupBy("user_id").agg(...)
agg_features_2 = base_features.groupBy("session_id").agg(...)

# Result: 40% reduction in compute time

d) Optimized Data Formats (Parquet with Compression)

# BEFORE: CSV files with no compression
raw_data.write.csv("gs://bucket/output/")

# AFTER: Parquet with Snappy compression
raw_data.write.parquet("gs://bucket/output/", compression="snappy")

# Result: 5x storage reduction, 3x faster reads

e) Dynamic Partition Overwrite (Avoid Full Table Rewrites)

# BEFORE: Overwrite entire table daily
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "static")
output.write.mode("overwrite").partitionBy("event_date").parquet("...")

# AFTER: Overwrite only affected partitions
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
output.write.mode("overwrite").partitionBy("event_date").parquet("...")

# Result: 80% reduction in write time for incremental updates

f) Cluster Autoscaling & Resource Right-Sizing

# Configured Dataproc clusters with:
# - Autoscaling: 5-50 workers based on pending tasks
# - Preemptible VMs: 70% of workers (cost saving)
# - Right-sized machine types: n1-highmem-4 instead of n1-standard-8

# Result: 40% reduction in compute costs

Cost Impact Breakdown:

Optimization	Cost Reduction
Partition pruning	15%
Broadcast joins	8%
Parquet + compression	20%
Autoscaling + preemptibles	40%
Caching + fewer recomputes	12%
Net Cost Reduction	~25%

Note: Percentages don’t sum linearly due to overlapping benefits.

---

3. Semi-Supervised Learning for SMS Spam Detection

Problem: Only 2% of SMS data labeled by users. Cold-start problem for new spam categories (e.g., COVID scams, cryptocurrency fraud).

Solution: Semi-supervised learning with DeBERTaV3 fine-tuning.

Approach:

Step 1: Self-Training with Pseudo-Labeling

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

class SemiSupervisedSpamClassifier:
    def __init__(self, model_name="microsoft/deberta-v3-base"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name, num_labels=2
        )

    def train_with_pseudo_labels(self, labeled_data, unlabeled_data, iterations=3):
        for iteration in range(iterations):
            # Step 1: Train on labeled data
            self.fine_tune(labeled_data)

            # Step 2: Predict on unlabeled data
            pseudo_labels = self.predict_with_confidence(unlabeled_data)

            # Step 3: Add high-confidence predictions to training set
            high_conf_samples = pseudo_labels[pseudo_labels['confidence'] > 0.95]
            labeled_data = pd.concat([labeled_data, high_conf_samples])

            print(f"Iteration {iteration+1}: Added {len(high_conf_samples)} pseudo-labeled samples")

    def predict_with_confidence(self, texts):
        self.model.eval()
        predictions = []

        for text in texts:
            inputs = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
            with torch.no_grad():
                outputs = self.model(**inputs)
                probs = torch.softmax(outputs.logits, dim=1)
                confidence, pred_label = torch.max(probs, dim=1)

            predictions.append({
                'text': text,
                'label': pred_label.item(),
                'confidence': confidence.item()
            })

        return pd.DataFrame(predictions)

Step 2: Data Augmentation for Rare Classes

import nlpaug.augmenter.word as naw

def augment_minority_class(spam_samples, num_augmented=5):
    augmenter = naw.SynonymAug(aug_src='wordnet')
    augmented_samples = []

    for text in spam_samples:
        for _ in range(num_augmented):
            augmented_text = augmenter.augment(text)
            augmented_samples.append(augmented_text)

    return augmented_samples

Step 3: Active Learning for Labeling Budget Optimization

def select_samples_for_labeling(model, unlabeled_data, budget=1000):
    """
    Select most informative samples using uncertainty sampling
    """
    predictions = model.predict_with_confidence(unlabeled_data)

    # Select samples with predictions closest to 0.5 (high uncertainty)
    predictions['uncertainty'] = 1 - abs(predictions['confidence'] - 0.5) * 2
    most_uncertain = predictions.nlargest(budget, 'uncertainty')

    return most_uncertain

Results:

Metric	Supervised Only	Semi-Supervised
Precision	0.82	0.89
Recall	0.76	0.85
F1-Score	0.79	0.87
Labeled Data Required	100% (200K)	10% (20K) + 180K unlabeled

Impact:

• 10% improvement in SMS spam detection (F1: 0.79 → 0.87)
• Reduced labeling costs by 80% through pseudo-labeling
• Faster adaptation to new spam types (2 days vs. 2 weeks)

---

The Impact: Production-Grade MLOps at Scale

Technical Achievements

1. Cost Optimization: 25% reduction in cloud infrastructure costs (~$100K+ annual savings)
2. Performance Improvement: Fraud detection rate up 8%, spam detection up 10%
3. Automation: Model retraining fully automated, reducing manual effort by 90%
4. Scalability: Pipelines handle 10B+ rows efficiently with sub-hour SLAs
5. Reliability: 99.9% uptime for ML serving endpoints

Business Impact

1. Customer Trust: Improved fraud detection protects millions of users daily
2. User Experience: Reduced spam exposure improves app engagement
3. Operational Efficiency: Data science team focuses on innovation, not maintenance
4. Competitive Advantage: Faster adaptation to evolving fraud/spam patterns

Learnings & Best Practices

What Worked:

• Drift detection over scheduled retraining: Data-driven triggers are more efficient
• Spark optimization compounds: Small improvements across many jobs add up
• Semi-supervised learning is practical: Leverage unlabeled data intelligently
• MLFlow standardization: Reproducibility and version control are non-negotiable

What Was Challenging:

• Tuning drift detection thresholds (false positives vs. false negatives)
• Balancing preemptible VM savings vs. job reliability
• Cold-start problem for entirely new fraud patterns (still requires some labeled data)

What I’d Do Differently:

• Implement feature store earlier (reduce redundant feature computation)
• Add more comprehensive data quality checks in pipelines
• Experiment with online learning for even faster adaptation

---

Technical Deep Dive: Spark Optimization Workflow

For engineers working with large-scale Spark pipelines:

1. Profiling & Identifying Bottlenecks

# Enable Spark UI and analyze execution plans
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

# Analyze query plan
df.explain("extended")

# Look for:
# - Full table scans (TableScan without partition filters)
# - Shuffle operations (Exchange)
# - Large broadcast joins (BroadcastExchange with >1GB data)

2. Partitioning Strategy

# Partition data by frequently filtered columns
transactions.write.partitionBy("event_date", "country") \
    .parquet("gs://bucket/transactions/")

# For high-cardinality columns, use bucketing
user_features.write.bucketBy(100, "user_id") \
    .sortBy("user_id") \
    .saveAsTable("user_features_bucketed")

3. Handling Skew

from pyspark.sql.functions import rand

# Salting technique for skewed joins
skewed_df = large_df.withColumn("salt", (rand() * 10).cast("int"))
small_df_salted = small_df.crossJoin(spark.range(10).withColumnRenamed("id", "salt"))

result = skewed_df.join(small_df_salted, ["join_key", "salt"]).drop("salt")

---

Conclusion

The Truecaller project demonstrates that MLOps is not just about automation—it’s about intelligent automation. By combining data drift detection, Spark optimization, and semi-supervised learning, we built a system that:

1. Adapts automatically to changing data patterns
2. Operates efficiently at billion-row scale
3. Delivers measurable business impact (cost reduction + performance improvement)

The key insight: Production ML is 20% modeling and 80% infrastructure, data engineering, and monitoring.

---

Technologies Used: Python, PySpark, MLFlow, Google Cloud Platform (BigQuery, Dataproc, Pub/Sub, Cloud Functions), DeBERTaV3, Transformers, XGBoost, LightGBM, Optuna, Great Expectations

← Back to Portfolio