Reddit Data Pipeline Guide: Building End-to-End ETL Systems for Analytics

import praw
import json
import boto3
from datetime import datetime, timedelta
from typing import Iterator, Dict, Any, Optional
from dataclasses import dataclass, asdict
import hashlib
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class RedditPost:
    """Reddit post data structure"""
    id: str
    subreddit: str
    title: str
    selftext: str
    author: str
    score: int
    num_comments: int
    created_utc: float
    url: str
    permalink: str
    is_self: bool
    over_18: bool
    upvote_ratio: float
    extracted_at: str

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


class RedditExtractor:
    """
    Extract Reddit data with incremental loading support

    Features:
    - Respects API rate limits
    - Tracks extraction state for incremental loads
    - Stores raw data to S3 for replay capability
    """

    def __init__(self,
                 client_id: str,
                 client_secret: str,
                 user_agent: str,
                 s3_bucket: str,
                 state_prefix: str = 'etl/state'):

        self.reddit = praw.Reddit(
            client_id=client_id,
            client_secret=client_secret,
            user_agent=user_agent
        )

        self.s3 = boto3.client('s3')
        self.bucket = s3_bucket
        self.state_prefix = state_prefix

    def _get_state(self, subreddit: str) -> Optional[Dict]:
        """Get extraction state for a subreddit"""
        key = f"{self.state_prefix}/{subreddit}.json"

        try:
            response = self.s3.get_object(Bucket=self.bucket, Key=key)
            return json.loads(response['Body'].read())
        except self.s3.exceptions.NoSuchKey:
            return None

    def _save_state(self, subreddit: str, state: Dict):
        """Save extraction state"""
        key = f"{self.state_prefix}/{subreddit}.json"

        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(state),
            ContentType='application/json'
        )

    def _convert_submission(self, submission) -> RedditPost:
        """Convert PRAW submission to our data structure"""
        return RedditPost(
            id=submission.id,
            subreddit=submission.subreddit.display_name,
            title=submission.title,
            selftext=submission.selftext or '',
            author=str(submission.author) if submission.author else '[deleted]',
            score=submission.score,
            num_comments=submission.num_comments,
            created_utc=submission.created_utc,
            url=submission.url,
            permalink=f"https://reddit.com{submission.permalink}",
            is_self=submission.is_self,
            over_18=submission.over_18,
            upvote_ratio=submission.upvote_ratio,
            extracted_at=datetime.utcnow().isoformat()
        )

    def extract_subreddit(self,
                          subreddit_name: str,
                          limit: int = 1000,
                          time_filter: str = 'day',
                          incremental: bool = True) -> Iterator[RedditPost]:
        """
        Extract posts from a subreddit

        Args:
            subreddit_name: Name of subreddit to extract
            limit: Maximum posts to fetch
            time_filter: Time filter for top posts
            incremental: Only fetch posts newer than last run
        """

        subreddit = self.reddit.subreddit(subreddit_name)
        state = self._get_state(subreddit_name) if incremental else None

        last_seen_utc = state.get('last_seen_utc', 0) if state else 0
        new_last_seen = last_seen_utc
        posts_extracted = 0

        logger.info(f"Extracting r/{subreddit_name}, last_seen: {last_seen_utc}")

        for submission in subreddit.top(time_filter=time_filter, limit=limit):
            # Skip already processed posts
            if incremental and submission.created_utc <= last_seen_utc:
                continue

            post = self._convert_submission(submission)
            posts_extracted += 1

            # Track newest post
            if submission.created_utc > new_last_seen:
                new_last_seen = submission.created_utc

            yield post

        # Update state
        if incremental and new_last_seen > last_seen_utc:
            self._save_state(subreddit_name, {
                'last_seen_utc': new_last_seen,
                'last_run': datetime.utcnow().isoformat(),
                'posts_extracted': posts_extracted
            })

        logger.info(f"Extracted {posts_extracted} posts from r/{subreddit_name}")

    def extract_multiple(self,
                         subreddits: list[str],
                         **kwargs) -> Iterator[RedditPost]:
        """Extract from multiple subreddits"""
        for subreddit in subreddits:
            yield from self.extract_subreddit(subreddit, **kwargs)

import json
import gzip
from datetime import datetime
from typing import List
import boto3

class RawDataStorage:
    """
    Store raw Reddit data with partitioning for efficient querying

    Data layout:
    s3://bucket/raw/reddit/posts/year=2025/month=01/day=15/hour=10/data.json.gz
    """

    def __init__(self, bucket: str, prefix: str = 'raw/reddit'):
        self.s3 = boto3.client('s3')
        self.bucket = bucket
        self.prefix = prefix

    def _generate_key(self, data_type: str, timestamp: datetime) -> str:
        """Generate partitioned S3 key"""
        return (
            f"{self.prefix}/{data_type}/"
            f"year={timestamp.year}/"
            f"month={timestamp.month:02d}/"
            f"day={timestamp.day:02d}/"
            f"hour={timestamp.hour:02d}/"
            f"data_{timestamp.strftime('%Y%m%d_%H%M%S')}.json.gz"
        )

    def store_batch(self,
                     posts: List[RedditPost],
                     data_type: str = 'posts') -> str:
        """
        Store a batch of posts to S3

        Args:
            posts: List of Reddit posts
            data_type: Type of data (posts, comments)

        Returns:
            S3 key where data was stored
        """
        if not posts:
            return None

        timestamp = datetime.utcnow()
        key = self._generate_key(data_type, timestamp)

        # Convert to JSONL and compress
        data = '\n'.join(json.dumps(post.to_dict()) for post in posts)
        compressed = gzip.compress(data.encode('utf-8'))

        # Upload to S3
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=compressed,
            ContentType='application/gzip',
            Metadata={
                'record_count': str(len(posts)),
                'extracted_at': timestamp.isoformat()
            }
        )

        logger.info(f"Stored {len(posts)} records to {key}")
        return key

    def list_partitions(self,
                        data_type: str,
                        start_date: datetime,
                        end_date: datetime) -> List[str]:
        """List S3 keys for a date range"""

        keys = []
        paginator = self.s3.get_paginator('list_objects_v2')

        prefix = f"{self.prefix}/{data_type}/"

        for page in paginator.paginate(Bucket=self.bucket, Prefix=prefix):
            for obj in page.get('Contents', []):
                # Parse partition info from key
                try:
                    parts = obj['Key'].split('/')
                    year = int(parts[-5].split('=')[1])
                    month = int(parts[-4].split('=')[1])
                    day = int(parts[-3].split('=')[1])

                    file_date = datetime(year, month, day)

                    if start_date <= file_date <= end_date:
                        keys.append(obj['Key'])
                except (IndexError, ValueError):
                    continue

        return keys

import pandas as pd
import numpy as np
from typing import List, Dict, Any
import re
from datetime import datetime
from transformers import pipeline
import hashlib

class RedditTransformer:
    """
    Transform raw Reddit data for analytics

    Transformations:
    - Data cleaning and validation
    - Text preprocessing
    - Feature engineering
    - Sentiment analysis
    - Deduplication
    """

    def __init__(self, enable_sentiment: bool = True):
        self.enable_sentiment = enable_sentiment

        if enable_sentiment:
            self.sentiment_analyzer = pipeline(
                'sentiment-analysis',
                model='cardiffnlp/twitter-roberta-base-sentiment-latest',
                device=-1
            )

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

        # Remove URLs
        text = re.sub(r'https?://\S+', '[URL]', text)

        # Remove Reddit-specific formatting
        text = re.sub(r'/r/\w+', '[SUBREDDIT]', text)
        text = re.sub(r'/u/\w+', '[USER]', text)

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

        # Remove markdown formatting
        text = re.sub(r'\*+([^*]+)\*+', r'\1', text)
        text = re.sub(r'#+\s*', '', text)

        return text

    def extract_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Engineer features from raw data"""

        df = df.copy()

        # Time-based features
        df['created_datetime'] = pd.to_datetime(df['created_utc'], unit='s')
        df['hour_of_day'] = df['created_datetime'].dt.hour
        df['day_of_week'] = df['created_datetime'].dt.dayofweek
        df['is_weekend'] = df['day_of_week'].isin([5, 6])

        # Text features
        df['title_length'] = df['title'].str.len()
        df['title_word_count'] = df['title'].str.split().str.len()
        df['content_length'] = df['selftext_clean'].str.len()
        df['content_word_count'] = df['selftext_clean'].str.split().str.len().fillna(0)

        # Question detection
        df['is_question'] = df['title'].str.contains(r'\?', regex=True)

        # Engagement metrics
        df['engagement_ratio'] = df['num_comments'] / (df['score'] + 1)
        df['controversy_score'] = (1 - df['upvote_ratio']) * df['score']

        # Content type classification
        df['content_type'] = np.where(
            df['is_self'],
            'text',
            np.where(
                df['url'].str.contains(r'\.(jpg|png|gif)', regex=True, na=False),
                'image',
                np.where(
                    df['url'].str.contains(r'(youtube|youtu\.be|v\.redd\.it)', regex=True, na=False),
                    'video',
                    'link'
                )
            )
        )

        return df

    def add_sentiment(self, df: pd.DataFrame, batch_size: int = 32) -> pd.DataFrame:
        """Add sentiment scores using transformer model"""

        if not self.enable_sentiment:
            df['sentiment'] = None
            df['sentiment_score'] = None
            return df

        df = df.copy()

        # Prepare text for analysis
        texts = (df['title'] + ' ' + df['selftext_clean'].fillna('')).tolist()
        texts = [t[:512] for t in texts]  # Model limit

        # Batch sentiment analysis
        sentiments = []
        scores = []

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

            for result in results:
                sentiments.append(result['label'].lower())
                scores.append(result['score'])

        df['sentiment'] = sentiments
        df['sentiment_score'] = scores

        return df

    def deduplicate(self, df: pd.DataFrame) -> pd.DataFrame:
        """Remove duplicate posts"""

        # Create content hash for fuzzy matching
        df['content_hash'] = (
            df['title'].str.lower() +
            df['selftext_clean'].fillna('').str.lower()
        ).apply(lambda x: hashlib.md5(x.encode()).hexdigest())

        # Keep first occurrence (usually the original)
        df = df.drop_duplicates(subset=['content_hash'], keep='first')

        # Also dedupe by ID
        df = df.drop_duplicates(subset=['id'], keep='first')

        return df

    def transform(self, posts: List[Dict]) -> pd.DataFrame:
        """
        Run full transformation pipeline

        Args:
            posts: List of raw post dictionaries

        Returns:
            Transformed DataFrame ready for loading
        """

        df = pd.DataFrame(posts)

        logger.info(f"Transforming {len(df)} posts")

        # Clean text
        df['title_clean'] = df['title'].apply(self.clean_text)
        df['selftext_clean'] = df['selftext'].apply(self.clean_text)

        # Engineer features
        df = self.extract_features(df)

        # Add sentiment
        df = self.add_sentiment(df)

        # Deduplicate
        initial_count = len(df)
        df = self.deduplicate(df)
        logger.info(f"Removed {initial_count - len(df)} duplicates")

        # Add processing metadata
        df['processed_at'] = datetime.utcnow().isoformat()
        df['pipeline_version'] = '1.0.0'

        return df

from google.cloud import bigquery
import pandas as pd
from typing import List, Dict
from datetime import datetime

class BigQueryLoader:
    """
    Load Reddit data into BigQuery with merge/upsert support

    Features:
    - Idempotent loads (safe to rerun)
    - Schema management
    - Partitioning by date
    - Clustering for query performance
    """

    SCHEMA = [
        bigquery.SchemaField('id', 'STRING', mode='REQUIRED'),
        bigquery.SchemaField('subreddit', 'STRING'),
        bigquery.SchemaField('title', 'STRING'),
        bigquery.SchemaField('title_clean', 'STRING'),
        bigquery.SchemaField('selftext', 'STRING'),
        bigquery.SchemaField('selftext_clean', 'STRING'),
        bigquery.SchemaField('author', 'STRING'),
        bigquery.SchemaField('score', 'INTEGER'),
        bigquery.SchemaField('num_comments', 'INTEGER'),
        bigquery.SchemaField('created_utc', 'FLOAT'),
        bigquery.SchemaField('created_datetime', 'TIMESTAMP'),
        bigquery.SchemaField('url', 'STRING'),
        bigquery.SchemaField('permalink', 'STRING'),
        bigquery.SchemaField('is_self', 'BOOLEAN'),
        bigquery.SchemaField('over_18', 'BOOLEAN'),
        bigquery.SchemaField('upvote_ratio', 'FLOAT'),
        bigquery.SchemaField('sentiment', 'STRING'),
        bigquery.SchemaField('sentiment_score', 'FLOAT'),
        bigquery.SchemaField('content_type', 'STRING'),
        bigquery.SchemaField('is_question', 'BOOLEAN'),
        bigquery.SchemaField('engagement_ratio', 'FLOAT'),
        bigquery.SchemaField('controversy_score', 'FLOAT'),
        bigquery.SchemaField('hour_of_day', 'INTEGER'),
        bigquery.SchemaField('day_of_week', 'INTEGER'),
        bigquery.SchemaField('is_weekend', 'BOOLEAN'),
        bigquery.SchemaField('extracted_at', 'TIMESTAMP'),
        bigquery.SchemaField('processed_at', 'TIMESTAMP'),
    ]

    def __init__(self,
                 project_id: str,
                 dataset_id: str,
                 table_id: str = 'reddit_posts'):

        self.client = bigquery.Client(project=project_id)
        self.project_id = project_id
        self.dataset_id = dataset_id
        self.table_id = table_id
        self.table_ref = f"{project_id}.{dataset_id}.{table_id}"

    def create_table_if_not_exists(self):
        """Create table with partitioning and clustering"""

        table = bigquery.Table(self.table_ref, schema=self.SCHEMA)

        # Partition by date for efficient time-based queries
        table.time_partitioning = bigquery.TimePartitioning(
            type_=bigquery.TimePartitioningType.DAY,
            field='created_datetime'
        )

        # Cluster by commonly filtered columns
        table.clustering_fields = ['subreddit', 'sentiment', 'content_type']

        try:
            self.client.create_table(table)
            logger.info(f"Created table {self.table_ref}")
        except Exception as e:
            if 'Already Exists' in str(e):
                logger.info(f"Table {self.table_ref} already exists")
            else:
                raise

    def load_dataframe(self, df: pd.DataFrame, write_mode: str = 'merge'):
        """
        Load DataFrame to BigQuery

        Args:
            df: Transformed DataFrame
            write_mode: 'append', 'replace', or 'merge' (upsert)
        """

        if df.empty:
            logger.info("Empty DataFrame, skipping load")
            return

        self.create_table_if_not_exists()

        if write_mode == 'merge':
            self._merge_load(df)
        elif write_mode == 'append':
            self._append_load(df)
        elif write_mode == 'replace':
            self._replace_load(df)

    def _merge_load(self, df: pd.DataFrame):
        """Merge/upsert data using staging table"""

        staging_table = f"{self.table_ref}_staging_{datetime.now().strftime('%Y%m%d%H%M%S')}"

        # Load to staging
        job_config = bigquery.LoadJobConfig(
            schema=self.SCHEMA,
            write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE
        )

        job = self.client.load_table_from_dataframe(
            df, staging_table, job_config=job_config
        )
        job.result()

        # Merge into main table
        merge_query = f"""
        MERGE `{self.table_ref}` T
        USING `{staging_table}` S
        ON T.id = S.id
        WHEN MATCHED THEN
            UPDATE SET
                score = S.score,
                num_comments = S.num_comments,
                upvote_ratio = S.upvote_ratio,
                engagement_ratio = S.engagement_ratio,
                controversy_score = S.controversy_score,
                processed_at = S.processed_at
        WHEN NOT MATCHED THEN
            INSERT ROW
        """

        self.client.query(merge_query).result()

        # Clean up staging
        self.client.delete_table(staging_table)

        logger.info(f"Merged {len(df)} records into {self.table_ref}")

    def _append_load(self, df: pd.DataFrame):
        """Append data (no deduplication)"""
        job_config = bigquery.LoadJobConfig(
            schema=self.SCHEMA,
            write_disposition=bigquery.WriteDisposition.WRITE_APPEND
        )

        job = self.client.load_table_from_dataframe(
            df, self.table_ref, job_config=job_config
        )
        job.result()

        logger.info(f"Appended {len(df)} records to {self.table_ref}")

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago
from airflow.models import Variable
from datetime import timedelta
import json

# Import our ETL components
from extractors.reddit_extractor import RedditExtractor
from storage.raw_storage import RawDataStorage
from transformers.reddit_transformer import RedditTransformer
from loaders.warehouse_loader import BigQueryLoader

# DAG default arguments
default_args = {
    'owner': 'data-engineering',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'email': ['data-alerts@company.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'execution_timeout': timedelta(hours=1),
}

# Configuration
CONFIG = {
    'subreddits': ['technology', 'programming', 'startups', 'entrepreneur'],
    's3_bucket': Variable.get('reddit_s3_bucket'),
    'bq_project': Variable.get('gcp_project_id'),
    'bq_dataset': 'reddit_analytics',
}


def extract_reddit_data(**context):
    """Extract data from Reddit API"""

    extractor = RedditExtractor(
        client_id=Variable.get('reddit_client_id'),
        client_secret=Variable.get('reddit_client_secret', deserialize_json=True),
        user_agent='DataPipeline/1.0',
        s3_bucket=CONFIG['s3_bucket']
    )

    storage = RawDataStorage(CONFIG['s3_bucket'])

    # Extract from each subreddit
    all_posts = []

    for subreddit in CONFIG['subreddits']:
        posts = list(extractor.extract_subreddit(
            subreddit,
            limit=500,
            time_filter='day',
            incremental=True
        ))
        all_posts.extend(posts)

    # Store raw data
    if all_posts:
        s3_key = storage.store_batch(all_posts)
        context['task_instance'].xcom_push(key='raw_s3_key', value=s3_key)
        context['task_instance'].xcom_push(key='post_count', value=len(all_posts))

    return len(all_posts)


def transform_reddit_data(**context):
    """Transform extracted data"""

    # Get raw data location from extract step
    ti = context['task_instance']
    s3_key = ti.xcom_pull(task_ids='extract', key='raw_s3_key')

    if not s3_key:
        logger.info("No data to transform")
        return 0

    # Load raw data from S3
    storage = RawDataStorage(CONFIG['s3_bucket'])
    # ... load data from s3_key

    # Transform
    transformer = RedditTransformer(enable_sentiment=True)
    # df = transformer.transform(raw_posts)

    # Save transformed data to intermediate location
    # ... save to S3 or temp location

    ti.xcom_push(key='transformed_count', value=0)  # placeholder


def load_to_warehouse(**context):
    """Load transformed data to BigQuery"""

    loader = BigQueryLoader(
        project_id=CONFIG['bq_project'],
        dataset_id=CONFIG['bq_dataset']
    )

    # Load transformed data
    # loader.load_dataframe(df, write_mode='merge')

    return "Load complete"


def validate_data_quality(**context):
    """Run data quality checks"""

    # Example quality checks
    checks = {
        'null_check': "SELECT COUNT(*) FROM table WHERE id IS NULL",
        'duplicate_check': "SELECT COUNT(*) - COUNT(DISTINCT id) FROM table",
        'freshness_check': "SELECT MAX(created_datetime) FROM table"
    }

    # Run checks and alert on failures
    pass


# Define DAG
with DAG(
    dag_id='reddit_etl_pipeline',
    default_args=default_args,
    description='ETL pipeline for Reddit data',
    schedule_interval='0 */4 * * *',  # Every 4 hours
    start_date=days_ago(1),
    catchup=False,
    tags=['reddit', 'etl', 'analytics'],
) as dag:

    start = DummyOperator(task_id='start')

    extract = PythonOperator(
        task_id='extract',
        python_callable=extract_reddit_data,
        provide_context=True,
    )

    transform = PythonOperator(
        task_id='transform',
        python_callable=transform_reddit_data,
        provide_context=True,
    )

    load = PythonOperator(
        task_id='load',
        python_callable=load_to_warehouse,
        provide_context=True,
    )

    validate = PythonOperator(
        task_id='validate',
        python_callable=validate_data_quality,
        provide_context=True,
    )

    end = DummyOperator(task_id='end')

    # Define task dependencies
    start >> extract >> transform >> load >> validate >> end

from abc import ABC, abstractmethod
from typing import List, Dict, Any
from dataclasses import dataclass
from enum import Enum
import pandas as pd

class CheckSeverity(Enum):
    WARNING = 'warning'
    ERROR = 'error'
    CRITICAL = 'critical'

@dataclass
class QualityCheckResult:
    check_name: str
    passed: bool
    severity: CheckSeverity
    message: str
    details: Dict[str, Any] = None


class DataQualityCheck(ABC):
    """Base class for data quality checks"""

    @abstractmethod
    def run(self, df: pd.DataFrame) -> QualityCheckResult:
        pass


class CompletenessCheck(DataQualityCheck):
    """Check for missing values"""

    def __init__(self, column: str, max_null_pct: float = 0.05):
        self.column = column
        self.max_null_pct = max_null_pct

    def run(self, df: pd.DataFrame) -> QualityCheckResult:
        null_count = df[self.column].isnull().sum()
        null_pct = null_count / len(df)

        passed = null_pct <= self.max_null_pct

        return QualityCheckResult(
            check_name=f"completeness_{self.column}",
            passed=passed,
            severity=CheckSeverity.ERROR if not passed else CheckSeverity.WARNING,
            message=f"Column {self.column}: {null_pct:.2%} null values",
            details={'null_count': null_count, 'null_pct': null_pct}
        )


class UniquenessCheck(DataQualityCheck):
    """Check for duplicate values"""

    def __init__(self, columns: List[str]):
        self.columns = columns

    def run(self, df: pd.DataFrame) -> QualityCheckResult:
        duplicate_count = df.duplicated(subset=self.columns).sum()
        duplicate_pct = duplicate_count / len(df)

        passed = duplicate_count == 0

        return QualityCheckResult(
            check_name=f"uniqueness_{'_'.join(self.columns)}",
            passed=passed,
            severity=CheckSeverity.ERROR if not passed else CheckSeverity.WARNING,
            message=f"Found {duplicate_count} duplicate rows",
            details={'duplicate_count': duplicate_count}
        )


class FreshnessCheck(DataQualityCheck):
    """Check data freshness"""

    def __init__(self, timestamp_column: str, max_age_hours: int = 24):
        self.timestamp_column = timestamp_column
        self.max_age_hours = max_age_hours

    def run(self, df: pd.DataFrame) -> QualityCheckResult:
        from datetime import datetime, timedelta

        max_timestamp = pd.to_datetime(df[self.timestamp_column]).max()
        age_hours = (datetime.utcnow() - max_timestamp).total_seconds() / 3600

        passed = age_hours <= self.max_age_hours

        return QualityCheckResult(
            check_name="freshness",
            passed=passed,
            severity=CheckSeverity.WARNING if not passed else CheckSeverity.WARNING,
            message=f"Data age: {age_hours:.1f} hours",
            details={'max_timestamp': str(max_timestamp), 'age_hours': age_hours}
        )


class DataQualityRunner:
    """Run all quality checks and aggregate results"""

    def __init__(self):
        self.checks: List[DataQualityCheck] = []

    def add_check(self, check: DataQualityCheck):
        self.checks.append(check)
        return self

    def run_all(self, df: pd.DataFrame) -> Dict:
        """Run all checks and return summary"""

        results = [check.run(df) for check in self.checks]

        return {
            'total_checks': len(results),
            'passed': sum(1 for r in results if r.passed),
            'failed': sum(1 for r in results if not r.passed),
            'critical_failures': [r for r in results if not r.passed and r.severity == CheckSeverity.CRITICAL],
            'results': results
        }