How We Process 2.4M Social Media Posts Per Hour

Scrapers (47 workers)
    ↓
Kafka (24 partitions, 7-day retention)
    ↓
NLP Workers (80 GPU instances)
    ↓  
TimescaleDB (primary storage)
    ↓
Redis (caching + real-time queries)
    ↓
API Layer (read-only, public facing)

# Kafka topic configuration
topic: social_media_posts
partitions: 24
replication_factor: 3
retention: 7 days (168 hours)
compression: lz4

# Producer settings (scrapers)
acks: 1  # Leader acknowledge, don't wait for replicas
compression_type: lz4
batch_size: 100KB
linger_ms: 50  # Wait up to 50ms to batch messages

# Consumer settings (NLP workers)
group_id: nlp_workers
auto_offset_reset: earliest
max_poll_records: 500

import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer

class NLPWorker:
    def __init__(self):
        # Load models once at startup
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

        # Sentiment model (DistilBERT)
        self.sentiment_model = AutoModelForSequenceClassification.from_pretrained(
            "distilbert-election-sentiment"
        ).to(self.device)

        # NER model (SpaCy)
        self.ner = spacy.load("en_election_ner")

        # Language detector (FastText)
        self.lang_detector = fasttext.load_model("lid.176.bin")

    def process_batch(self, posts):
        """Process a batch of posts"""
        results = []

        for post in posts:
            # 1. Detect language (3ms)
            lang = self.lang_detector.predict(post.text)[0][0]

            if lang not in ['en', 'es', 'zh']:
                continue  # Skip non-target languages

            # 2. Extract entities (12ms)
            doc = self.ner(post.text)
            entities = {
                'people': [e.text for e in doc.ents if e.label_ == 'PERSON'],
                'orgs': [e.text for e in doc.ents if e.label_ == 'ORG'],
                'locations': [e.text for e in doc.ents if e.label_ == 'GPE']
            }

            # 3. Sentiment classification (45ms)
            inputs = self.tokenizer(post.text, return_tensors="pt", truncation=True, max_length=512)
            inputs = {k: v.to(self.device) for k, v in inputs.items()}

            with torch.no_grad():
                outputs = self.sentiment_model(**inputs)
                probs = torch.softmax(outputs.logits, dim=1)[0]

            sentiment = {
                'positive': probs[0].item(),
                'neutral': probs[1].item(),
                'negative': probs[2].item()
            }

            results.append({
                'post_id': post.id,
                'language': lang,
                'entities': entities,
                'sentiment': sentiment,
                'processed_at': time.time()
            })

        return results

CREATE TABLE posts (
    time            TIMESTAMPTZ NOT NULL,
    platform        TEXT NOT NULL,
    post_id         TEXT NOT NULL,
    author_id       TEXT,
    text            TEXT,
    language        TEXT,
    sentiment_pos   FLOAT,
    sentiment_neg   FLOAT,
    sentiment_neu   FLOAT,
    entities        JSONB,
    embedding       vector(768),  -- Sentence embedding for similarity search

    PRIMARY KEY (time, post_id)
);

-- Create hypertable (automatic time-based partitioning)
SELECT create_hypertable('posts', 'time', chunk_time_interval => INTERVAL '1 day');

-- Indexes
CREATE INDEX idx_platform ON posts (platform, time DESC);
CREATE INDEX idx_author ON posts (author_id, time DESC);
CREATE INDEX idx_entities ON posts USING GIN (entities);
CREATE INDEX idx_embedding ON posts USING ivfflat (embedding vector_cosine_ops);

-- Compression (automatically compress data older than 7 days)
ALTER TABLE posts SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'platform',
    timescaledb.compress_orderby = 'time DESC'
);

SELECT add_compression_policy('posts', INTERVAL '7 days');

# NLP worker writes to both PostgreSQL and Redis
def store_results(results):
    # Write to PostgreSQL (durable storage)
    db.execute("INSERT INTO posts VALUES (...)", results)

    # Write to Redis (fast queries)
    for result in results:
        # Sorted set by timestamp
        redis.zadd(f"posts:{result.platform}", {result.post_id: result.time})

        # Hash for post details
        redis.hset(f"post:{result.post_id}", mapping=result)

        # Expire after 24 hours
        redis.expire(f"post:{result.post_id}", 86400)

        # Update real-time counters
        redis.hincrby("stats:posts_by_platform", result.platform, 1)
        redis.hincrby("stats:posts_by_sentiment", result.sentiment_label, 1)

from datasketch import MinHash, MinHashLSH

# Initialize LSH index
lsh = MinHashLSH(threshold=0.85, num_perm=128)

def check_duplicate(post):
    # Create MinHash signature
    m = MinHash(num_perm=128)
    for word in post.text.lower().split():
        m.update(word.encode('utf8'))

    # Check for similar posts
    similar = lsh.query(m)

    if similar:
        # Duplicate found, store reference only
        return {
            'is_duplicate': True,
            'original_id': similar[0],
            'similarity': calculate_jaccard(post.text, get_text(similar[0]))
        }
    else:
        # New post, add to index
        lsh.insert(post.id, m)
        return {'is_duplicate': False}

# MinHash LSH is O(1) for lookups, handles 2.4M posts/hour easily

# Example metrics exposed by NLP workers
posts_processed_total{platform="twitter"} 1847293
posts_processed_total{platform="reddit"} 892847

processing_duration_seconds{quantile="0.5"} 0.058
processing_duration_seconds{quantile="0.95"} 0.143
processing_duration_seconds{quantile="0.99"} 0.287

model_inference_duration_seconds{model="sentiment"} 0.045
model_inference_duration_seconds{model="ner"} 0.012

kafka_consumer_lag{partition="0"} 142
kafka_consumer_lag{partition="1"} 89