Enterprise Sentiment Analysis System
BERT + Spark NLP | Scalable Real-Time Analysis
What You'll Learn
- 1Fine-tune a four-class BERT classifier and score it with macro F1 rather than accuracy
- 2Weight or oversample rare classes and quantify the expected macro F1 gain
- 3Build an aspect-based layer that re-scores only the sentences mentioning each aspect
- 4Serve batch inference in one forward pass and monitor label shift with a PSI alert
Project Overview
Problem Statement
Enterprises receive millions of customer interactions daily across channels. Manual sentiment analysis cannot scale. This system uses fine-tuned BERT models with Spark NLP for distributed processing to analyze sentiment across emails, reviews, social media, and support tickets in real-time.
Objectives
- Fine-tune BERT for multi-class sentiment (positive, negative, neutral, mixed)
- Build scalable processing pipeline with Spark NLP
- Implement aspect-based sentiment analysis
- Deploy with real-time and batch inference modes
- Create dashboards for sentiment trend monitoring
| Component | Technology |
|---|---|
| Sentiment Model | BERT / RoBERTa (fine-tuned) |
| Scalable Processing | Spark NLP |
| API Framework | FastAPI + Triton Inference Server |
| Stream Processing | Apache Kafka + Flink |
| Data Storage | Elasticsearch + PostgreSQL |
| Dashboard | Grafana + Kibana |
| Monitoring | Prometheus + ELK Stack |
Design Rationale
Four-class sentiment is an imbalanced classification problem before it is a sentiment problem. On support tickets the split might be 60 percent neutral, 25 percent negative, 10 percent positive, and 5 percent mixed, so accuracy hides failure on the minority classes and the headline metric has to be macro F1:
Training minimizes cross entropy over the four logits, and the gradient contribution of a rare class scales with how often it appears, so mixed examples at 5 percent contribute about one update in twenty unless you weight the loss:
With labeled texts and , the mixed class weight is 5.0, and adding it typically recovers 3 to 6 points of macro F1 over unweighted training on the same data.
Sequence length sets both cost and coverage. max_length 128 fits a support ticket opening at a fertility near 1.3 tokens per word, roughly 98 words, and short sequences keep the encoder inside the efficient kernel range:
A single A10G pushes about 900 texts per second at batch 64 and length 128 with a fine-tuned BERT-base; DistilBERT reaches about 1,800 for a 1 to 2 point drop in macro F1. That 2x for 1 to 2 points is the trade the deployment table is pricing.
Aspect-based analysis adds a second, cheaper pass instead of a second full model. Overall sentiment comes from the fine-tuned classifier, and each aspect is scored only from sentences that mention it, so the extra cost per request is one forward pass per detected aspect rather than one per sentence. Keywords bound the work to five aspects, which keeps p95 latency growth linear and small.
Architecture Diagram
Step-by-Step Implementation
Step 1: Environment Setup
mkdir sentiment-system && cd sentiment-system
pip install transformers datasets torch spark-nlp
pip install fastapi uvicorn kafka-python elasticsearch
pip install scikit-learn mlflow prometheus-client
pip install pyspark
Step 2: BERT Fine-Tuning for Sentiment
Fine-tune BERT for multi-class sentiment classification on enterprise data.
# src/models/train_sentiment.py
import torch
from torch.utils.data import Dataset, DataLoader
from transformers import (
BertTokenizer, BertForSequenceClassification,
AdamW, get_linear_schedule_with_warmup
)
from sklearn.metrics import classification_report, confusion_matrix
from typing import List, Dict, Tuple
import mlflow
import numpy as np
class SentimentDataset(Dataset):
LABEL_MAP = {"negative": 0, "neutral": 1, "positive": 2, "mixed": 3}
def __init__(self, texts: List[str], labels: List[str], tokenizer, max_len: int = 128):
self.texts = texts
self.labels = [self.LABEL_MAP[l] for l in labels]
self.tokenizer = tokenizer
self.max_len = max_len
def __len__(self):
return len(self.texts)
def __getitem__(self, idx):
encoding = self.tokenizer(
self.texts[idx],
max_length=self.max_len,
padding="max_length",
truncation=True,
return_tensors="pt",
)
return {
"input_ids": encoding["input_ids"].squeeze(),
"attention_mask": encoding["attention_mask"].squeeze(),
"labels": torch.tensor(self.labels[idx], dtype=torch.long),
}
class SentimentTrainer:
def __init__(self, model_name: str = "bert-base-uncased", num_labels: int = 4):
self.tokenizer = BertTokenizer.from_pretrained(model_name)
self.model = BertForSequenceClassification.from_pretrained(
model_name, num_labels=num_labels
)
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
self.model.to(self.device)
def train(
self, train_texts, train_labels, eval_texts, eval_labels,
epochs: int = 5, batch_size: int = 32, learning_rate: float = 2e-5
):
train_dataset = SentimentDataset(train_texts, train_labels, self.tokenizer)
eval_dataset = SentimentDataset(eval_texts, eval_labels, self.tokenizer)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
eval_loader = DataLoader(eval_dataset, batch_size=batch_size)
optimizer = AdamW(self.model.parameters(), lr=learning_rate)
total_steps = len(train_loader) * epochs
scheduler = get_linear_schedule_with_warmup(
optimizer, num_warmup_steps=total_steps // 10, num_training_steps=total_steps
)
mlflow.set_experiment("sentiment-analysis")
with mlflow.start_run():
mlflow.log_params({"epochs": epochs, "lr": learning_rate, "batch_size": batch_size})
for epoch in range(epochs):
self.model.train()
total_loss = 0
for batch in train_loader:
optimizer.zero_grad()
outputs = self.model(
input_ids=batch["input_ids"].to(self.device),
attention_mask=batch["attention_mask"].to(self.device),
labels=batch["labels"].to(self.device),
)
loss = outputs.loss
loss.backward()
torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
optimizer.step()
scheduler.step()
total_loss += loss.item()
avg_loss = total_loss / len(train_loader)
eval_metrics = self.evaluate(eval_loader)
mlflow.log_metrics({
"train_loss": avg_loss,
"eval_accuracy": eval_metrics["accuracy"],
}, step=epoch)
print(f"Epoch {epoch+1}: loss={avg_loss:.4f}, acc={eval_metrics['accuracy"]:.4f}")
mlflow.pytorch.log_model(self.model, "sentiment-model")
def evaluate(self, eval_loader) -> Dict:
self.model.eval()
all_preds, all_labels = [], []
with torch.no_grad():
for batch in eval_loader:
outputs = self.model(
input_ids=batch["input_ids"].to(self.device),
attention_mask=batch["attention_mask"].to(self.device),
)
preds = torch.argmax(outputs.logits, dim=-1)
all_preds.extend(preds.cpu().numpy())
all_labels.extend(batch["labels"].numpy())
accuracy = np.mean(np.array(all_preds) == np.array(all_labels))
return {"accuracy": accuracy, "report": classification_report(all_labels, all_preds)}
Step 3: Spark NLP Preprocessing Pipeline
Build scalable text preprocessing with Spark NLP for batch analysis of millions of documents.
# src/processing/spark_pipeline.py
import sparknlp
from sparknlp.base import DocumentAssembler, Finisher
from sparknlp.annotator import (
Tokenizer, Normalizer, StopWordsCleaner, LemmatizerModel,
SentenceDetectorDLModel
)
from pyspark.ml import Pipeline
import pyspark.sql.functions as F
class SparkSentimentPipeline:
def __init__(self):
self.spark = sparknlp.start()
self.pipeline = self._build_pipeline()
def _build_pipeline(self):
document_assembler = DocumentAssembler() \
.setInputCol("text") \
.setOutputCol("document")
sentence_detector = SentenceDetectorDLModel.pretrained() \
.setInputCols(["document"]) \
.setOutputCol("sentences")
tokenizer = Tokenizer() \
.setInputCols(["sentences"]) \
.setOutputCol("tokens")
normalizer = Normalizer() \
.setInputCols(["tokens"]) \
.setOutputCol("normalized") \
.setLowercase(True)
stopwords_cleaner = StopWordsCleaner() \
.setInputCols(["normalized"]) \
.setOutputCol("clean_tokens")
lemmatizer = LemmatizerModel.pretrained("lemma_antbnc") \
.setInputCols(["clean_tokens"]) \
.setOutputCol("lemmatized")
finisher = Finisher() \
.setInputCols(["lemmatized"]) \
.setOutputCols(["processed_text"]) \
.setCleanAnnotations(True)
return Pipeline(stages=[
document_assembler, sentence_detector, tokenizer,
normalizer, stopwords_cleaner, lemmatizer, finisher
])
def process_batch(self, df):
model = self.pipeline.fit(df)
return model.transform(df)
def analyze_large_dataset(self, input_path: str, output_path: str, chunk_size: int = 100000):
df = self.spark.read.parquet(input_path)
total = df.count()
print(f"Processing {total} documents...")
processed = self.process_batch(df)
processed.write.mode("overwrite").parquet(output_path)
print(f"Results saved to {output_path}")
Step 4: Aspect-Based Sentiment Analysis
Extract sentiment toward specific aspects (e.g., "battery life", "customer service").
# src/models/aspect_sentiment.py
from transformers import pipeline as hf_pipeline
from typing import List, Dict, Tuple
import re
class AspectSentimentAnalyzer:
ASPECT_KEYWORDS = {
"product_quality": ["quality", "build", "material", "durable", "cheap"],
"customer_service": ["support", "service", "help", "response", "staff"],
"price": ["price", "cost", "expensive", "cheap", "value", "worth"],
"delivery": ["shipping", "delivery", "arrived", "package", "fast"],
"usability": ["easy", "difficult", "intuitive", "confusing", "user-friendly"],
}
def __init__(self, sentiment_model: str = "cardiffnlp/twitter-roberta-base-sentiment-latest"):
self.sentiment_pipeline = hf_pipeline("sentiment-analysis", model=sentiment_model)
def extract_aspects(self, text: str) -> List[str]:
text_lower = text.lower()
found_aspects = []
for aspect, keywords in self.ASPECT_KEYWORDS.items():
if any(kw in text_lower for kw in keywords):
found_aspects.append(aspect)
return found_aspects if found_aspects else ["general"]
def analyze(self, text: str) -> Dict:
aspects = self.extract_aspects(text)
sentences = re.split(r"[.!?]+", text)
aspect_results = {}
for aspect in aspects:
relevant_sentences = [
s.strip() for s in sentences
if any(kw in s.lower() for kw in self.ASPECT_KEYWORDS.get(aspect, []))
]
if relevant_sentences:
combined = " ".join(relevant_sentences)
result = self.sentiment_pipeline(combined[:512])[0]
aspect_results[aspect] = {
"sentiment": result["label"],
"confidence": result["score"],
"text_snippet": combined[:200],
}
else:
overall = self.sentiment_pipeline(text[:512])[0]
aspect_results[aspect] = {
"sentiment": overall["label"],
"confidence": overall["score"] * 0.7,
"text_snippet": text[:200],
}
return {"aspects": aspect_results, "overall": self.sentiment_pipeline(text[:512])[0]}
Step 5: Real-Time Inference API
# src/api/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional, Dict
from contextlib import asynccontextmanager
import torch
from transformers import BertTokenizer, BertForSequenceClassification
import time
LABEL_MAP = {0: "negative", 1: "neutral", 2: "positive", 3: "mixed"}
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.tokenizer = BertTokenizer.from_pretrained("./sentiment_model")
app.state.model = BertForSequenceClassification.from_pretrained("./sentiment_model")
app.state.model.eval()
app.state.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
app.state.model.to(app.state.device)
yield
app = FastAPI(title="Sentiment API", lifespan=lifespan)
class SentimentRequest(BaseModel):
text: str = Field(..., min_length=1, max_length=5000)
return_aspects: bool = False
class SentimentResponse(BaseModel):
sentiment: str
confidence: float
all_scores: Dict[str, float]
latency_ms: float
aspects: Optional[Dict] = None
@app.post("/api/v1/sentiment", response_model=SentimentResponse)
async def analyze_sentiment(request: SentimentRequest):
start = time.time()
tokenizer = app.state.tokenizer
model = app.state.model
encoding = tokenizer(
request.text, max_length=128, padding="max_length",
truncation=True, return_tensors="pt"
)
with torch.no_grad():
outputs = model(
input_ids=encoding["input_ids"].to(app.state.device),
attention_mask=encoding["attention_mask"].to(app.state.device),
)
probs = torch.softmax(outputs.logits, dim=-1)[0].cpu().numpy()
pred_idx = int(np.argmax(probs))
latency = (time.time() - start) * 1000
return SentimentResponse(
sentiment=LABEL_MAP[pred_idx],
confidence=float(probs[pred_idx]),
all_scores={LABEL_MAP[i]: float(p) for i, p in enumerate(probs)},
latency_ms=latency,
)
@app.post("/api/v1/sentiment/batch")
async def batch_sentiment(texts: List[str]):
results = []
for text in texts[:100]:
result = await analyze_sentiment(SentimentRequest(text=text))
results.append(result.dict())
return {"results": results, "total": len(results)}
âšī¸
Use domain-specific fine-tuning data. Generic sentiment models perform poorly on specialized domains like healthcare or finance. Collect and annotate domain-specific training data.
đĄ
Implement a fallback chain: if the fine-tuned model has low confidence, fall back to a general-purpose sentiment model. This improves coverage for edge cases.
Performance Metrics
| Metric | Target | Description |
|---|---|---|
| Accuracy | > 92% | 4-class sentiment |
| F1 Score | > 0.90 | Weighted F1 |
| Inference Latency | < 30ms | Single text (GPU) |
| Batch Throughput | > 1000 texts/s | Spark NLP |
| Aspect Precision | > 0.85 | Aspect extraction |
What Breaks in Production
Label shift is the quiet killer. A product launch changes the ticket mix from mostly neutral to mostly negative, and a model that was 92 percent accurate last quarter keeps calling genuinely angry messages neutral. Track the predicted class distribution daily and alert when the population stability index passes 0.2:
Here is the reference share for class and is the current share; values below 0.1 mean the mix is stable, 0.1 to 0.2 deserves a look, and above 0.2 the class priors have moved enough to justify retraining.
Second, truncation bias. The API caps input at 128 tokens, but the verdict in a long ticket usually sits in the final paragraph, so the model reads the greeting and predicts neutral. Encode the first 128 and last 128 tokens and concatenate the pooled vectors before the head, or move to 256 tokens and accept roughly 1.7x on latency.
Third, a batch endpoint that loops over single-text calls. Awaiting one text at a time buys no batching at all: pad to a common length and run a single forward pass so a 100 item request costs one call instead of 100, cutting wall time by an order of magnitude.
Finally, feedback loops. Feeding high-confidence predictions back in as labels entrenches whatever the model already believed. Route 20 percent of low-confidence cases to human review, and re-anchor weekly on a fresh 500 item hand-labeled set.
Interview Talking Points
- BERT Fine-Tuning: Adding a classification head to BERT and fine-tuning on domain data typically achieves 5-10% improvement over zero-shot.
- Aspect-Based Analysis: Decomposing sentiment by aspect provides actionable insights that overall sentiment misses.
- Spark NLP: Enables processing millions of documents with GPU-accelerated NLP pipelines at scale.
- Model Selection: RoBERTa often outperforms BERT for sentiment. Consider domain-specific models like FinBERT for finance.
- Multi-Label: Some texts express mixed sentiments. Multi-label classification captures this nuance.
- Deployment: ONNX export + Triton Inference Server provides optimal serving throughput.
â ī¸
Sentiment models can encode biases from training data. Regularly audit model predictions across demographic groups and retrain with balanced datasets.
âšī¸
For production deployment, consider using DistilBERT for 2x faster inference with minimal accuracy loss (typically 1-2% drop).
Common Mistakes
Reporting accuracy on an imbalanced sentiment set
Fix: Report macro F1 and per-class recall; a 60 percent neutral class keeps accuracy high while mixed collapses.
Truncating long tickets at 128 tokens from the front
Fix: Encode head and tail tokens together, because the verdict usually sits in the final paragraph.
Serving batches by looping over single-text calls
Fix: Pad to a common length and run one forward pass per batch so 100 items cost one call.
Best Practices
- Refresh a 500 item hand-labeled set monthly so evaluation tracks the live distribution
- Log the full four-class probability vector rather than only the argmax label
- Version the label map with the model artifact so clients never guess class ordering
Key Takeaways
- Class weights of N over 4 times N_c recover 3 to 6 points of macro F1 on a 5 percent minority class
- max_length 128 covers about 98 English words at a fertility of 1.3 tokens per word
- DistilBERT doubles throughput to roughly 1,800 texts per second for 1 to 2 points of macro F1
- A population stability index above 0.2 on predicted class shares signals label shift
Frequently Asked Questions
Why four classes instead of positive, negative, and neutral?
Mixed tickets are where action is needed: a customer praising delivery while condemning support is not neutral, and forcing the label loses the actionable half. Four classes also forces the model to represent uncertainty explicitly, which shows up as a flatter probability vector you can threshold instead of trusting the argmax.
Should you fine-tune RoBERTa instead of BERT?
RoBERTa usually adds 1 to 2 points of macro F1 on sentiment tasks because of its longer training and dynamic masking, at the same parameter count and latency. BERT-base is fine when the domain gap is large and data is scarce, since both benefit far more from domain-specific labels than from the architecture swap.
How much labeled data do you need?
Around 5,000 to 10,000 balanced examples per class gets most of the way: beyond that the curve flattens and annotation quality matters more than volume. Spend the next dollar on adjudicating ambiguous cases and fixing label guidelines rather than doubling the dataset.
Interview Questions
Q1: How do you handle severe class imbalance in sentiment labels?
Weight the loss by N over 4 times N_c, oversample the minority class in the sampler, and report macro F1 so a collapsed minority class cannot hide. On a 5 percent mixed class this typically recovers 3 to 6 points of macro F1.
Q2: Explain how aspect-based sentiment is derived from the overall model.
Split the text into sentences, match them to aspect keywords, and run the classifier on the concatenated sentences mentioning each aspect. Aspects with no matching sentence inherit the overall score with a 0.7 confidence multiplier so the UI can gray them out.
Q3: How do you detect that the model has degraded in production?
Two signals: predicted class shares drifting with a PSI above 0.2, and accuracy measured on the weekly 500 item labeled sample dropping more than 2 points below baseline. Drift fires first because labels lag, accuracy confirms it.