Build production NLP systems that extract meaning from text. Covers text preprocessing, tokenization strategies, named entity recognition, sentiment analysis, text classification, and the patterns that turn unstructured text into actionable structured data.
Natural Language Processing Pipelines are end-to-end systems that transform raw text into structured, actionable outputs—such as sentiment scores, named entities, topic labels, or machine-generated summaries—through a sequence of well-defined, reusable stages. They matter when text data is the primary signal in a system: in customer support workflows, real-time content moderation, recommendation engines, or enterprise search. Pipelines are essential when processing thousands of documents per minute with consistent, reproducible results across varied input formats and domains.
A pipeline is a series of processing stages, each consuming and producing structured data. Each stage may be a standalone service, a batch job, or a streaming process. The output of one stage becomes the input of the next, forming a data flow that is both modular and composable.
The typical input is a JSON object with a text field and optional metadata:
{
"id": "doc-12345",
"text": "The new model was released last week and has shown remarkable performance in low-resource settings.",
"source": "blog",
"timestamp": "2024-04-05T12:34:56Z"
}Each stage must accept and emit a consistent schema. A common output schema for a pipeline is:
{
"id": "doc-12345",
"text": "...",
"timestamp": "...",
"annotations": [
{
"type": "entity",
"label": "Model",
"start": 15,
"end": 22,
"text": "model"
},
{
"type": "sentiment",
"score": 0.87,
"label": "positive"
}
],
"metadata": {
"pipeline_version": "v2.3",
"processing_time_ms": 327
}
}The pipeline is composed by defining a sequence of processors, each with a unique name and configuration. The configuration is typically a YAML file describing the order, parameters, and dependencies.
stages:
- name: tokenizer
type: "sentencepiece"
model: "models/sentencepiece.model"
max_length: 512
add_special_tokens: true
- name: embedding
type: "transformer"
model: "distilbert-base-uncased"
batch_size: 16
device: "cuda"
- name: ner
type: "crf"
model: "models/ner-crf.pt"
label_map: "data/labels.json"
- name: sentiment
type: "linear"
model: "models/sentiment-mlp.pkl"
threshold: 0.5The pipeline engine reads the configuration and orchestrates execution. It may run in batch mode, streaming mode, or as a microservice.
In batch mode, pipelines process large volumes of data offline. The workflow uses standard tools: git, docker, psql, curl, and systemctl.
A batch pipeline starts with a docker-compose.yml to orchestrate services:
# docker-compose.yaml
version: "3.8"
services:
pipeline:
image: "nlp-pipeline:latest"
volumes:
- ./data/input:/input
- ./data/output:/output
depends_on:
- postgres
command: >
/app/run_pipeline.py
--config /app/pipeline.yaml
--input /input
--output /output
--workers 4
--batch-size 32
postgres:
image: "postgres:15"
environment:
POSTGRES_DB: "nlp"
POSTGRES_USER: "nlp"
POSTGRES_PASSWORD: "secret"
volumes:
- postgres_data:/var/lib/postgresql/data
volumes:
postgres_data:The entrypoint script, run_pipeline.py, uses click for CLI and pandas for data handling:
# run_pipeline.py
import click
import pandas as pd
from pathlib import Path
@click.command()
@click.option("--config", type=click.Path(exists=True), required=True)
@click.option("--input", type=click.Path(exists=True), required=True)
@click.option("--output", type=click.Path(), required=True)
@click.option("--workers", type=int, default=1)
@click.option("--batch-size", type=int, default=16)
def run_pipeline(config, input, output, workers, batch_size):
# Load config
with open(config) as f:
config_data = yaml.safe_load(f)
# Read input files
input_path = Path(input)
input_files = list(input_path.glob("*.json"))
# Process in batches
for batch in chunked(input_files, batch_size):
docs = []
for file in batch:
with open(file) as f:
docs.append(json.load(f))
# Apply pipeline stages
for stage in config_data["stages"]:
if stage["type"] == "sentencepiece":
docs = tokenize_with_sentencepiece(docs, stage)
elif stage["type"] == "transformer":
docs = embed_with_transformer(docs, stage)
elif stage["type"] == "crf":
docs = annotate_with_crf(docs, stage)
# Save output
output_path = Path(output)
output_path.mkdir(exist_ok=True)
for doc in docs:
with open(output_path / f"{doc['id']}.json", "w") as f:
json.dump(doc, f, indent=2)
click.echo(f"Processed {len(input_files)} documents.")
if __name__ == "__main__":
run_pipeline()The pipeline runs as a systemctl service:
# /etc/systemd/system/nlp-pipeline.service
[Unit]
Description=NLP Pipeline Batch Processor
After=network.target
[Service]
Type=oneshot
User=app
WorkingDirectory=/opt/nlp
ExecStart=/usr/bin/python /opt/nlp/run_pipeline.py --config /opt/nlp/pipeline.yaml --input /data/input --output /data/output
StandardOutput=journal
StandardError=journal
Environment=PYTHONPATH=/opt/nlp
[Install]
WantedBy=multi-user.targetFor real-time systems, pipelines consume a stream of documents via Apache Kafka. The pipeline is a consumer that processes incoming messages and emits results to an output topic.
The consumer uses kafka-python and confluent-kafka:
# stream_processor.py
from confluent_kafka import Consumer, Producer
import json
import threading
class NLPStreamProcessor:
def __init__(self, config_path, bootstrap_servers):
self.config = self.load_config(config_path)
self.consumer = Consumer({
"bootstrap.servers": bootstrap_servers,
"group.id": "nlp-pipeline",
"auto.offset.reset": "earliest"
})
self.producer = Producer({"bootstrap.servers": bootstrap_servers})
def load_config(self, path):
with open(path) as f:
return yaml.safe_load(f)
def consume_and_process(self):
self.consumer.subscribe(["raw-text"])
while True:
msg = self.consumer.poll(timeout=1.0)
if msg is None:
continue
if msg.error():
print(f"Consumer error: {msg.error()}")
continue
raw_doc = json.loads(msg.value().decode("utf-8"))
processed_doc = self.process_document(raw_doc)
self.producer.produce(
topic="processed-text",
key=processed_doc["id"],
value=json.dumps(processed_doc),
callback=self.delivery_callback
)
self.producer.flush()
def process_document(self, doc):
for stage in self.config["stages"]:
if stage["type"] == "transformer":
doc = self.embed_with_transformer(doc, stage)
elif stage["type"] == "crf":
doc = self.annotate_with_crf(doc, stage)
elif stage["type"] == "linear":
doc = self.predict_sentiment(doc, stage)
return doc
def delivery_callback(self, err, msg):
if err:
print(f"Message delivery failed: {err}")
else:
print(f"Message delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")
# Run in background
if __name__ == "__main__":
processor = NLPStreamProcessor("pipeline.yaml", "kafka:9092")
thread = threading.Thread(target=processor.consume_and_process)
thread.start()The pipeline runs as a Docker container with systemctl:
# /etc/systemd/system/nlp-stream.service
[Unit]
Description=NLP Stream Processor
After=network.target
[Service]
Type=simple
User=app
WorkingDirectory=/opt/nlp
ExecStart=/usr/bin/python /opt/nlp/stream_processor.py /opt/nlp/pipeline.yaml kafka:9092
Restart=always
RestartSec=30
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.targetConfiguration is central to pipeline reliability. Misconfigurations silently degrade performance or break processing.
A common failure is incorrect model paths in stage configuration. If the transformer stage expects a model at models/bert-base-uncased, but the file is named bert-base-uncased.tar.gz, the pipeline logs FileNotFoundError but continues without error:
[ERROR] Failed to load model at models/bert-base-uncased: No such file or directory
[WARNING] Using fallback model: models/bert-base-uncased/model.binThe pipeline should validate configuration at startup:
def validate_configuration(config):
for stage in config["stages"]:
if stage["type"] == "transformer":
model_path = Path(stage["model"])
if not model_path.exists():
raise FileNotFoundError(f"Transformer model not found: {stage['model']}")
elif stage["type"] == "crf":
if "label_map" not in stage:
raise KeyError("CRF stage missing required 'label_map' field")A subtle but frequent error is mismatched input and output schemas. A stage expects {"text": "..."} but receives {"content": "..."}. The pipeline must log a warning:
{
"error": "Schema mismatch",
"stage": "embedding",
"expected": ["text"],
"actual": ["content", "title"],
"input": {
"content": "The model was released...",
"title": "New Release"
}
}Pipelines evolve. Versioning ensures reproducibility. Each pipeline is deployed as a Docker image tagged with a version.
docker build -t nlp-pipeline:v2.3 .
docker tag nlp-pipeline:v2.3 registry.example.com/nlp-pipeline:v2.3
docker push registry.example.com/nlp-pipeline:v2.3Deployment uses kubectl to manage pods:
# kubectl apply -f pipeline-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: nlp-pipeline
spec:
replicas: 3
selector:
matchLabels:
app: nlp-pipeline
template:
metadata:
labels:
app: nlp-pipeline
spec:
containers:
- name: pipeline
image: registry.example.com/nlp-pipeline:v2.3
ports:
- containerPort: 8080
env:
- name: PIPELINE_CONFIG
value: "/app/pipeline.yaml"
volumeMounts:
- name: config
mountPath: /app/pipeline.yaml
- name: models
mountPath: /models
volumes:
- name: config
configMap:
name: pipeline-config
- name: models
persistentVolumeClaim:
claimName: nlp-modelsThe pipeline logs version and processing time in metadata:
{
"id": "doc-12345",
"text": "...",
"metadata": {
"pipeline_version": "v2.3",
"processing_time_ms": 214,
"input_schema": "v1",
"output_schema": "v2"
}
}The pipeline must emit metrics to prometheus via exporter endpoints.
# metrics.py
from prometheus_client import Counter, Histogram, start_http_server
REQUEST_COUNT = Counter("nlp_pipeline_requests_total", "Total requests processed", ["stage"])
PROCESSING_TIME = Histogram("nlp_pipeline_processing_time_seconds", "Processing time per document", ["stage"])
def record_metrics(stage_name, duration, success=True):
REQUEST_COUNT.labels(stage=stage_name).inc()
PROCESSING_TIME.labels(stage=stage_name).observe(duration)Failure modes include:
null text field causes the tokenizer stage to fail silently, producing null output.processed-text topic lags behind raw-text, the pipeline emits a warning and triggers a alert via alertmanager.The pipeline may also emit health checks:
GET /health
HTTP/1.1 200 OK
Content-Type: application/json
{
"status": "healthy",
"timestamp": "2024-04-05T13:20:00Z",
"pipeline_version": "v2.3",
"consumer_lag": 42,
"last_processing_time_ms": 312
}Natural Language Processing Pipelines are not just a sequence of models and scripts—they are systems designed for consistency, reliability, and scalability. From batch processing with docker and systemctl to real-time streaming with Kafka, the pipeline is a data flow that must be configured, versioned, monitored, and maintained. The key to success lies in precise schema definitions, robust error handling, and clear metrics. When a document enters the pipeline, it is not just transformed—it is tracked, validated, and trusted.
This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.
We use cookies for analytics (Google Analytics) and advertising (Google AdSense) to improve your experience and support free content. Privacy Policy