Reddit Data Pipeline Architecture Guide

import asyncio
import aiohttp
from dataclasses import dataclass
from typing import List, AsyncIterator
from datetime import datetime
import json

@dataclass
class RedditPost:
    id: str
    subreddit: str
    title: str
    selftext: str
    author: str
    score: int
    num_comments: int
    created_utc: float
    url: str
    permalink: str


class RedditIngestionService:
    """
    Async Reddit data ingestion with rate limiting.
    Supports multiple subreddits and pagination.
    """

    BASE_URL = "https://oauth.reddit.com"

    def __init__(
        self,
        client_id: str,
        client_secret: str,
        user_agent: str,
        requests_per_minute: int = 60
    ):
        self.client_id = client_id
        self.client_secret = client_secret
        self.user_agent = user_agent
        self.rate_limit = requests_per_minute
        self.request_times = []
        self.access_token = None
        self.session = None

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        await self._authenticate()
        return self

    async def __aexit__(self, *args):
        await self.session.close()

    async def _authenticate(self):
        """Get OAuth2 access token."""
        auth = aiohttp.BasicAuth(self.client_id, self.client_secret)
        data = {'grant_type': 'client_credentials'}

        async with self.session.post(
            'https://www.reddit.com/api/v1/access_token',
            auth=auth,
            data=data,
            headers={'User-Agent': self.user_agent}
        ) as resp:
            result = await resp.json()
            self.access_token = result['access_token']

    async def _rate_limit(self):
        """Enforce rate limiting."""
        now = datetime.now().timestamp()
        self.request_times = [
            t for t in self.request_times
            if now - t < 60
        ]

        if len(self.request_times) >= self.rate_limit:
            sleep_time = 60 - (now - self.request_times[0])
            await asyncio.sleep(sleep_time)

        self.request_times.append(now)

    async def fetch_subreddit(
        self,
        subreddit: str,
        sort: str = 'new',
        limit: int = 100,
        after: str = None
    ) -> List[RedditPost]:
        """Fetch posts from a subreddit."""
        await self._rate_limit()

        params = {'limit': limit}
        if after:
            params['after'] = after

        headers = {
            'Authorization': f'Bearer {self.access_token}',
            'User-Agent': self.user_agent
        }

        url = f'{self.BASE_URL}/r/{subreddit}/{sort}'

        async with self.session.get(url, params=params, headers=headers) as resp:
            data = await resp.json()

        posts = []
        for child in data['data']['children']:
            post_data = child['data']
            posts.append(RedditPost(
                id=post_data['id'],
                subreddit=post_data['subreddit'],
                title=post_data['title'],
                selftext=post_data.get('selftext', ''),
                author=post_data['author'],
                score=post_data['score'],
                num_comments=post_data['num_comments'],
                created_utc=post_data['created_utc'],
                url=post_data['url'],
                permalink=post_data['permalink']
            ))

        return posts, data['data'].get('after')

    async def stream_subreddits(
        self,
        subreddits: List[str],
        poll_interval: int = 30
    ) -> AsyncIterator[RedditPost]:
        """Continuously stream new posts from subreddits."""
        seen_ids = set()

        while True:
            for subreddit in subreddits:
                posts, _ = await self.fetch_subreddit(subreddit, limit=25)

                for post in posts:
                    if post.id not in seen_ids:
                        seen_ids.add(post.id)
                        yield post

            # Trim seen_ids to prevent memory growth
            if len(seen_ids) > 10000:
                seen_ids = set(list(seen_ids)[-5000:])

            await asyncio.sleep(poll_interval)


# Usage example
async def main():
    async with RedditIngestionService(
        client_id='your_client_id',
        client_secret='your_secret',
        user_agent='DataPipeline/1.0'
    ) as service:
        async for post in service.stream_subreddits(['technology', 'programming']):
            print(f"New post: {post.title[:50]}")
            # Send to message queue
            await publish_to_queue(post)

from kafka import KafkaProducer
from kafka.errors import KafkaError
import json
from dataclasses import asdict

class RedditKafkaProducer:
    """Kafka producer for Reddit posts with delivery guarantees."""

    def __init__(
        self,
        bootstrap_servers: str,
        topic: str = 'reddit-posts'
    ):
        self.topic = topic
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            key_serializer=lambda k: k.encode('utf-8'),
            acks='all',  # Wait for all replicas
            retries=3,
            max_in_flight_requests_per_connection=1,
            enable_idempotence=True  # Exactly-once semantics
        )

    def send_post(self, post: RedditPost):
        """Send post to Kafka with subreddit-based partitioning."""
        try:
            future = self.producer.send(
                self.topic,
                key=post.subreddit,  # Partition by subreddit
                value=asdict(post)
            )
            # Block for synchronous sends (optional)
            record_metadata = future.get(timeout=10)
            return True
        except KafkaError as e:
            print(f"Failed to send: {e}")
            return False

    def flush(self):
        """Ensure all messages are sent."""
        self.producer.flush()

    def close(self):
        self.producer.close()

import re
from datetime import datetime
from typing import Dict, Any, Optional
import html
import emoji

class RedditTransformer:
    """Transform raw Reddit data for analysis."""

    def __init__(self):
        self.url_pattern = re.compile(r'https?://\S+')
        self.mention_pattern = re.compile(r'u/\w+')
        self.subreddit_pattern = re.compile(r'r/\w+')

    def clean_text(self, text: str) -> str:
        """Clean and normalize Reddit text."""
        if not text:
            return ''

        # Decode HTML entities
        text = html.unescape(text)

        # Remove markdown formatting
        text = re.sub(r'\*\*(.*?)\*\*', r'\1', text)  # Bold
        text = re.sub(r'\*(.*?)\*', r'\1', text)  # Italic
        text = re.sub(r'~~(.*?)~~', r'\1', text)  # Strikethrough
        text = re.sub(r'\[([^\]]+)\]\([^\)]+\)', r'\1', text)  # Links

        # Normalize whitespace
        text = re.sub(r'\s+', ' ', text)

        return text.strip()

    def extract_features(self, post: Dict[str, Any]) -> Dict[str, Any]:
        """Extract analytical features from post."""
        text = f"{post.get('title', '')} {post.get('selftext', '')}"

        return {
            'word_count': len(text.split()),
            'char_count': len(text),
            'url_count': len(self.url_pattern.findall(text)),
            'mention_count': len(self.mention_pattern.findall(text)),
            'subreddit_refs': self.subreddit_pattern.findall(text),
            'has_emoji': bool(emoji.emoji_count(text)),
            'is_question': '?' in post.get('title', ''),
            'hour_of_day': datetime.fromtimestamp(
                post['created_utc']
            ).hour,
            'day_of_week': datetime.fromtimestamp(
                post['created_utc']
            ).weekday()
        }

    def transform(self, post: Dict[str, Any]) -> Dict[str, Any]:
        """Full transformation pipeline for a post."""
        transformed = {
            'id': post['id'],
            'subreddit': post['subreddit'].lower(),
            'title_clean': self.clean_text(post.get('title', '')),
            'body_clean': self.clean_text(post.get('selftext', '')),
            'author': post.get('author', '[deleted]'),
            'score': post.get('score', 0),
            'num_comments': post.get('num_comments', 0),
            'created_utc': post['created_utc'],
            'created_at': datetime.fromtimestamp(
                post['created_utc']
            ).isoformat(),
            'url': post.get('url', ''),
            'permalink': post.get('permalink', ''),
            'ingested_at': datetime.utcnow().isoformat()
        }

        # Add extracted features
        transformed['features'] = self.extract_features(post)

        return transformed

from typing import Dict, List, Any
from concurrent.futures import ThreadPoolExecutor
import spacy

class PostEnrichmentPipeline:
    """Enrich Reddit posts with NLP features."""

    def __init__(self, sentiment_model, batch_size: int = 32):
        self.sentiment_model = sentiment_model
        self.nlp = spacy.load('en_core_web_sm')
        self.batch_size = batch_size

    def extract_entities(self, text: str) -> Dict[str, List[str]]:
        """Extract named entities from text."""
        doc = self.nlp(text)
        entities = {}

        for ent in doc.ents:
            if ent.label_ not in entities:
                entities[ent.label_] = []
            entities[ent.label_].append(ent.text)

        return entities

    def batch_sentiment(self, texts: List[str]) -> List[Dict]:
        """Batch sentiment analysis for efficiency."""
        results = []

        for i in range(0, len(texts), self.batch_size):
            batch = texts[i:i + self.batch_size]
            batch_results = self.sentiment_model.predict(batch)
            results.extend(batch_results)

        return results

    def categorize_post(self, post: Dict) -> str:
        """Assign category based on content signals."""
        text = f"{post['title_clean']} {post['body_clean']}".lower()

        categories = {
            'question': ['how', 'what', 'why', 'when', '?'],
            'recommendation': ['recommend', 'suggest', 'best', 'looking for'],
            'complaint': ['issue', 'problem', 'broken', 'frustrated'],
            'review': ['review', 'experience', 'used', 'tried'],
            'discussion': ['thoughts', 'opinion', 'discuss']
        }

        for category, keywords in categories.items():
            if any(kw in text for kw in keywords):
                return category

        return 'general'

    def enrich_batch(self, posts: List[Dict]) -> List[Dict]:
        """Enrich a batch of posts."""
        # Extract texts for batch processing
        texts = [
            f"{p['title_clean']} {p['body_clean']}"
            for p in posts
        ]

        # Batch sentiment analysis
        sentiments = self.batch_sentiment(texts)

        # Parallel entity extraction
        with ThreadPoolExecutor(max_workers=4) as executor:
            entities = list(executor.map(self.extract_entities, texts))

        # Combine enrichments
        enriched = []
        for i, post in enumerate(posts):
            post['sentiment'] = sentiments[i]
            post['entities'] = entities[i]
            post['category'] = self.categorize_post(post)
            enriched.append(post)

        return enriched

-- Core posts table with partitioning by date
CREATE TABLE reddit_posts (
    id VARCHAR(20) PRIMARY KEY,
    subreddit VARCHAR(50) NOT NULL,
    title TEXT NOT NULL,
    body TEXT,
    author VARCHAR(50),
    score INTEGER DEFAULT 0,
    num_comments INTEGER DEFAULT 0,
    created_at TIMESTAMP NOT NULL,
    ingested_at TIMESTAMP DEFAULT NOW(),

    -- Enrichment fields
    sentiment_label VARCHAR(20),
    sentiment_score DECIMAL(5,4),
    category VARCHAR(30),

    -- Extracted features (JSONB for flexibility)
    features JSONB,
    entities JSONB
) PARTITION BY RANGE (created_at);

-- Create monthly partitions
CREATE TABLE reddit_posts_2026_01 PARTITION OF reddit_posts
    FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');

CREATE TABLE reddit_posts_2026_02 PARTITION OF reddit_posts
    FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');

-- Indexes for common queries
CREATE INDEX idx_posts_subreddit ON reddit_posts(subreddit);
CREATE INDEX idx_posts_created ON reddit_posts(created_at DESC);
CREATE INDEX idx_posts_sentiment ON reddit_posts(sentiment_label);
CREATE INDEX idx_posts_subreddit_date ON reddit_posts(subreddit, created_at DESC);

-- Full-text search index
CREATE INDEX idx_posts_fts ON reddit_posts
    USING GIN (to_tsvector('english', title || ' ' || COALESCE(body, '')));

-- JSONB index for entity queries
CREATE INDEX idx_posts_entities ON reddit_posts USING GIN (entities);

-- Aggregation table for analytics
CREATE TABLE subreddit_daily_stats (
    subreddit VARCHAR(50),
    date DATE,
    post_count INTEGER,
    avg_score DECIMAL(10,2),
    avg_comments DECIMAL(10,2),
    positive_ratio DECIMAL(5,4),
    negative_ratio DECIMAL(5,4),
    top_entities JSONB,
    PRIMARY KEY (subreddit, date)
);

-- Materialized view for trending analysis
CREATE MATERIALIZED VIEW trending_topics AS
SELECT
    subreddit,
    DATE_TRUNC('hour', created_at) as hour,
    COUNT(*) as post_count,
    AVG(score) as avg_score,
    jsonb_agg(DISTINCT entities->'ORG') as mentioned_orgs
FROM reddit_posts
WHERE created_at > NOW() - INTERVAL '24 hours'
GROUP BY subreddit, DATE_TRUNC('hour', created_at);

CREATE UNIQUE INDEX idx_trending_topics ON trending_topics(subreddit, hour);