Data Processing Pipeline

Omnivore's data processing pipeline transforms raw web content into structured, analyzable data. From HTML parsing to entity extraction, each stage is optimized for performance and accuracy.

Processing Pipeline Overview

Pipeline Stages

The processing pipeline consists of sequential stages, each optimized for specific tasks:

HTML Parsing

Parser Configuration

TOML
COPY
001[processing.parser]
002# Parser engine
003engine = "html5ever"  # Fast HTML5 parser
004
005# Parsing options
006fix_errors = true
007remove_scripts = true
008remove_styles = true
009decode_entities = true
010
011# Content extraction
012extract_text = true
013extract_links = true
014extract_images = true
015extract_metadata = true
016
017# Performance
018max_parse_size = 10485760  # 10MB
019timeout = 5000  # 5 seconds

Structured Data Extraction

Extract structured data from HTML using various methods:

RUST
COPY
001// Extract structured data
002let extractor = DataExtractor::new()
003    .extract_json_ld()      // JSON-LD structured data
004    .extract_microdata()    // Schema.org microdata
005    .extract_rdfa()         // RDFa annotations
006    .extract_opengraph()    // Open Graph metadata
007    .extract_twitter_card() // Twitter Card data
008    .build();
009
010let structured_data = extractor.extract(&html)?;

Content Cleaning

Boilerplate Removal

Omnivore uses advanced algorithms to identify and remove non-content elements:

• Navigation menus: Header and footer navigation
• Advertisements: Ad blocks and promotional content
• Sidebars: Related articles, widgets, social media
• Comments: User comments and discussion threads

TOML
COPY
001[processing.cleaning]
002# Cleaning algorithms
003algorithm = "readability"  # or "boilerpipe", "custom"
004
005# Content detection
006min_text_length = 100
007min_word_count = 50
008max_link_density = 0.3
009
010# Element removal
011remove_elements = ["script", "style", "iframe", "embed"]
012remove_classes = ["advertisement", "sidebar", "footer", "header"]
013remove_ids = ["comments", "related-posts", "social-share"]
014
015# Text cleaning
016normalize_whitespace = true
017remove_empty_lines = true
018fix_encoding = true

Main Content Extraction

BASH
COPY
001# Extract main content from HTML
002omnivore process extract-content page.html --output content.txt
003
004# With custom rules
005omnivore process extract-content page.html \
006  --algorithm readability \
007  --min-words 100 \
008  --output content.txt
009
010# Batch processing
011omnivore process extract-content ./html-files/ \
012  --output-dir ./clean-content/ \
013  --workers 10

Text Processing

Text Normalization

TOML
COPY
001[processing.text]
002# Language detection
003detect_language = true
004supported_languages = ["en", "es", "fr", "de", "zh"]
005
006# Text normalization
007lowercase = false  # Preserve case for NER
008remove_accents = false
009expand_contractions = true
010fix_unicode = true
011
012# Tokenization
013tokenizer = "spacy"  # or "nltk", "custom"
014preserve_entities = true
015preserve_urls = true
016preserve_emails = true

Language Processing

RUST
COPY
001use omnivore::processing::{LanguageDetector, TextProcessor};
002
003// Detect language
004let detector = LanguageDetector::new();
005let language = detector.detect(&text)?;
006
007// Process text based on language
008let processor = TextProcessor::for_language(language);
009let processed = processor
010    .tokenize()
011    .lemmatize()
012    .remove_stopwords()
013    .process(&text)?;

Metadata Extraction

Automatic Metadata Detection

Omnivore automatically extracts various types of metadata:

Metadata Extraction Rules

JSON
COPY
001{
002  "metadata_rules": {
003    "title": [
004      {"selector": "title"},
005      {"selector": "meta[property='og:title']", "attribute": "content"},
006      {"selector": "h1", "first": true}
007    ],
008    "author": [
009      {"selector": "meta[name='author']", "attribute": "content"},
010      {"selector": ".author-name"},
011      {"pattern": "By ([A-Z][a-z]+ [A-Z][a-z]+)"}
012    ],
013    "date": [
014      {"selector": "meta[property='article:published_time']", "attribute": "content"},
015      {"selector": "time", "attribute": "datetime"},
016      {"pattern": "\d{4}-\d{2}-\d{2}"}
017    ],
018    "category": [
019      {"selector": "meta[property='article:section']", "attribute": "content"},
020      {"selector": ".category"},
021      {"xpath": "//nav[@class='breadcrumb']/a[last()]"}
022    ]
023  }
024}

Content Enrichment

Entity Recognition

BASH
COPY
001# Extract entities from text
002omnivore process extract-entities content.txt \
003  --types person,org,location \
004  --output entities.json
005
006# With confidence threshold
007omnivore process extract-entities content.txt \
008  --min-confidence 0.8 \
009  --output entities.json
010
011# Custom NER model
012omnivore process extract-entities content.txt \
013  --model custom-ner-model.bin \
014  --output entities.json

Content Classification

TOML
COPY
001[processing.classification]
002# Classification model
003model = "fasttext"  # or "bert", "custom"
004model_path = "./models/classifier.bin"
005
006# Categories
007categories = [
008  "technology",
009  "business",
010  "science",
011  "health",
012  "sports",
013  "entertainment"
014]
015
016# Classification settings
017min_confidence = 0.7
018multi_label = true
019max_categories = 3

Sentiment Analysis

BASH
COPY
001# Analyze sentiment
002omnivore process analyze-sentiment content.txt
003
004# Output:
005# Overall: Positive (0.78)
006# Polarity: 0.65
007# Subjectivity: 0.42
008# Emotions: joy (0.45), trust (0.32)

Processing Pipelines

Custom Pipeline Configuration

YAML
COPY
001# pipeline.yaml
002name: news-processing
003description: Pipeline for news article processing
004
005stages:
006  - name: parse
007    type: html_parser
008    config:
009      engine: html5ever
010      extract_metadata: true
011
012  - name: clean
013    type: content_cleaner
014    config:
015      algorithm: readability
016      min_words: 100
017
018  - name: extract_entities
019    type: ner
020    config:
021      model: en_core_web_lg
022      types: [person, org, location]
023
024  - name: classify
025    type: classifier
026    config:
027      model: news-classifier
028      categories: [politics, tech, business]
029
030  - name: summarize
031    type: summarizer
032    config:
033      algorithm: extractive
034      sentences: 3
035
036  - name: store
037    type: storage
038    config:
039      backend: elasticsearch
040      index: news-articles

Pipeline Execution

BASH
COPY
001# Run custom pipeline
002omnivore process run-pipeline pipeline.yaml \
003  --input crawled-data.json \
004  --output processed-data.json
005
006# Run with monitoring
007omnivore process run-pipeline pipeline.yaml \
008  --input crawled-data.json \
009  --output processed-data.json \
010  --monitor \
011  --metrics-port 9090

Storage & Indexing

Storage Backends

TOML
COPY
001[processing.storage]
002# Primary storage
003backend = "postgresql"  # or "mongodb", "elasticsearch", "rocksdb"
004connection_string = "postgres://user:pass@localhost/omnivore"
005
006# Batch settings
007batch_size = 1000
008flush_interval = 5000  # ms
009
010# Compression
011compress = true
012compression_algorithm = "zstd"
013
014# Indexing
015create_indexes = true
016index_fields = ["title", "content", "url", "domain"]
017
018# Full-text search
019enable_fts = true
020fts_language = "english"

Search Index Creation

BASH
COPY
001# Create search index
002omnivore index create processed-data.json \
003  --backend elasticsearch \
004  --index omnivore-content
005
006# With custom mapping
007omnivore index create processed-data.json \
008  --backend elasticsearch \
009  --index omnivore-content \
010  --mapping custom-mapping.json
011
012# Update existing index
013omnivore index update omnivore-content \
014  --add-documents new-data.json

Batch Processing

Large-Scale Processing

BASH
COPY
001# Process large dataset in batches
002omnivore process batch \
003  --input s3://bucket/crawled-data/ \
004  --output s3://bucket/processed-data/ \
005  --batch-size 10000 \
006  --workers 20 \
007  --checkpoint ./checkpoints/
008
009# Resume from checkpoint
010omnivore process batch \
011  --resume ./checkpoints/batch-12345.json \
012  --output s3://bucket/processed-data/

Stream Processing

BASH
COPY
001# Process streaming data
002omnivore process stream \
003  --input kafka://localhost:9092/crawled-pages \
004  --output kafka://localhost:9092/processed-pages \
005  --pipeline news-pipeline.yaml \
006  --workers 10

Quality Control

Data Validation

TOML
COPY
001[processing.validation]
002# Validation rules
003validate_required_fields = ["title", "content", "url"]
004min_content_length = 100
005max_content_length = 1000000
006
007# Quality checks
008check_language = true
009expected_language = "en"
010check_encoding = true
011check_duplicates = true
012
013# Rejection criteria
014reject_on_error = false
015log_rejected = true
016rejected_output = "./rejected.json"

Processing Metrics

BASH
COPY
001# View processing statistics
002omnivore process stats
003
004# Output:
005# Documents processed: 45,231
006# Success rate: 98.5%
007# Average processing time: 125ms
008# Entities extracted: 234,521
009# Categories assigned: 89.3%
010# Errors: 679 (1.5%)

Performance Optimization

Optimization Strategies

• Parallel Processing: Use multiple workers for CPU-bound tasks
• Caching: Cache parsed HTML and extracted entities
• Batch Operations: Process documents in batches
• Lazy Loading: Load models on-demand
• GPU Acceleration: Use GPU for NLP models when available

TOML
COPY
001[processing.performance]
002# Parallel processing
003workers = 16
004thread_pool_size = 32
005
006# Caching
007enable_cache = true
008cache_backend = "redis"
009cache_ttl = 3600
010
011# Memory management
012max_memory = "8GB"
013gc_interval = 1000
014
015# GPU settings
016use_gpu = true
017gpu_device = 0
018gpu_batch_size = 32

Error Handling

Error Recovery

BASH
COPY
001# Process with error handling
002omnivore process run \
003  --input data.json \
004  --output processed.json \
005  --on-error continue \
006  --error-log errors.log \
007  --max-errors 100
008
009# Retry failed documents
010omnivore process retry \
011  --failed errors.log \
012  --output retry-processed.json \
013  --max-retries 3

Best Practices