Big Data Processing for Reddit: Spark, Distributed Computing, and Scale

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *

def create_spark_session(
    app_name: str = "RedditAnalysis",
    memory: str = "8g",
    cores: int = 4
) -> SparkSession:
    """
    Create optimized Spark session for Reddit data processing.
    """
    spark = (SparkSession.builder
        .appName(app_name)
        .config("spark.driver.memory", memory)
        .config("spark.executor.memory", memory)
        .config("spark.sql.shuffle.partitions", cores * 4)
        .config("spark.sql.adaptive.enabled", "true")
        .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        .getOrCreate())

    # Set log level
    spark.sparkContext.setLogLevel("WARN")

    return spark

# Define Reddit post schema for efficient parsing
REDDIT_POST_SCHEMA = StructType([
    StructField("id", StringType(), True),
    StructField("title", StringType(), True),
    StructField("selftext", StringType(), True),
    StructField("author", StringType(), True),
    StructField("subreddit", StringType(), True),
    StructField("score", IntegerType(), True),
    StructField("num_comments", IntegerType(), True),
    StructField("created_utc", LongType(), True),
    StructField("upvote_ratio", FloatType(), True),
    StructField("is_self", BooleanType(), True),
    StructField("over_18", BooleanType(), True)
])

# Usage
spark = create_spark_session()
print(f"Spark version: {spark.version}")

class RedditDataLoader:
    """Load and manage large Reddit datasets with Spark."""

    def __init__(self, spark: SparkSession):
        self.spark = spark

    def load_json(self, path: str, schema=None):
        """Load JSON files (line-delimited or standard)."""
        reader = self.spark.read

        if schema:
            reader = reader.schema(schema)

        return reader.json(path)

    def load_parquet(self, path: str):
        """Load Parquet files (recommended for large datasets)."""
        return self.spark.read.parquet(path)

    def load_compressed(self, path: str, compression: str = "zstd"):
        """Load compressed JSON (common for Reddit archives)."""
        return (self.spark.read
            .option("compression", compression)
            .json(path))

    def load_partitioned(self, base_path: str, partitions: list = None):
        """
        Load partitioned dataset (e.g., by date or subreddit).

        Args:
            base_path: Base path to partitioned data
            partitions: Specific partitions to load (e.g., ['year=2024', 'month=01'])
        """
        if partitions:
            paths = [f"{base_path}/{p}" for p in partitions]
            return self.spark.read.parquet(*paths)
        else:
            return self.spark.read.parquet(base_path)

    def optimize_dataframe(self, df, target_partitions: int = None):
        """Optimize DataFrame for processing."""
        if target_partitions:
            df = df.repartition(target_partitions)

        # Cache if will be reused
        # df = df.cache()

        return df

# Usage
loader = RedditDataLoader(spark)

# Load large dataset
posts_df = loader.load_json(
    "s3://reddit-data/posts/*.json.zst",
    schema=REDDIT_POST_SCHEMA
)

print(f"Loaded {posts_df.count():,} posts")

from pyspark.sql import functions as F
from pyspark.ml.feature import Tokenizer, StopWordsRemover, CountVectorizer
from pyspark.ml import Pipeline

class DistributedTextProcessor:
    """Process Reddit text at scale with Spark ML."""

    def __init__(self, spark: SparkSession):
        self.spark = spark

    def clean_text(self, df, text_col: str = "selftext"):
        """Clean Reddit text using Spark SQL functions."""
        return df.withColumn(
            "cleaned_text",
            F.lower(
                F.regexp_replace(
                    F.regexp_replace(
                        F.regexp_replace(
                            F.col(text_col),
                            r'https?://\S+', ''  # Remove URLs
                        ),
                        r'r/\w+', ''  # Remove subreddit mentions
                    ),
                    r'[^a-zA-Z\s]', ' '  # Remove special chars
                )
            )
        )

    def create_text_pipeline(self):
        """Create ML pipeline for text processing."""
        # Tokenize
        tokenizer = Tokenizer(inputCol="cleaned_text", outputCol="tokens")

        # Remove stop words
        remover = StopWordsRemover(inputCol="tokens", outputCol="filtered_tokens")

        # Vectorize
        vectorizer = CountVectorizer(
            inputCol="filtered_tokens",
            outputCol="features",
            vocabSize=10000,
            minDF=5
        )

        return Pipeline(stages=[tokenizer, remover, vectorizer])

    def compute_word_frequencies(self, df, text_col: str = "selftext"):
        """Compute word frequencies across entire corpus."""
        # Clean and tokenize
        cleaned = self.clean_text(df, text_col)

        # Explode words
        words_df = cleaned.select(
            F.explode(F.split(F.col("cleaned_text"), r'\s+')).alias("word")
        ).filter(F.length("word") > 2)

        # Count frequencies
        return (words_df
            .groupBy("word")
            .count()
            .orderBy(F.desc("count")))

# Usage
processor = DistributedTextProcessor(spark)
cleaned_df = processor.clean_text(posts_df)
word_freq = processor.compute_word_frequencies(posts_df)
word_freq.show(20)

class RedditAggregator:
    """Perform large-scale aggregations on Reddit data."""

    def __init__(self, spark: SparkSession):
        self.spark = spark

    def subreddit_stats(self, df):
        """Compute statistics per subreddit."""
        return (df
            .groupBy("subreddit")
            .agg(
                F.count("*").alias("post_count"),
                F.sum("score").alias("total_score"),
                F.avg("score").alias("avg_score"),
                F.stddev("score").alias("score_stddev"),
                F.sum("num_comments").alias("total_comments"),
                F.avg("num_comments").alias("avg_comments"),
                F.countDistinct("author").alias("unique_authors"),
                F.min("created_utc").alias("first_post"),
                F.max("created_utc").alias("last_post")
            )
            .orderBy(F.desc("post_count")))

    def time_series_aggregation(self, df, freq: str = "day"):
        """Aggregate by time period."""
        # Convert UTC timestamp to date
        df_with_date = df.withColumn(
            "date",
            F.to_date(F.from_unixtime("created_utc"))
        )

        if freq == "day":
            group_col = "date"
        elif freq == "week":
            df_with_date = df_with_date.withColumn(
                "week",
                F.date_trunc("week", F.col("date"))
            )
            group_col = "week"
        elif freq == "month":
            df_with_date = df_with_date.withColumn(
                "month",
                F.date_trunc("month", F.col("date"))
            )
            group_col = "month"

        return (df_with_date
            .groupBy(group_col)
            .agg(
                F.count("*").alias("post_count"),
                F.sum("score").alias("total_score"),
                F.avg("score").alias("avg_score"),
                F.sum("num_comments").alias("total_comments")
            )
            .orderBy(group_col))

    def author_activity(self, df, min_posts: int = 10):
        """Analyze author posting patterns."""
        author_stats = (df
            .filter(F.col("author") != "[deleted]")
            .groupBy("author")
            .agg(
                F.count("*").alias("post_count"),
                F.sum("score").alias("total_karma"),
                F.avg("score").alias("avg_score"),
                F.countDistinct("subreddit").alias("subreddits_active"),
                F.collect_set("subreddit").alias("subreddits")
            )
            .filter(F.col("post_count") >= min_posts))

        return author_stats

# Usage
aggregator = RedditAggregator(spark)
subreddit_stats = aggregator.subreddit_stats(posts_df)
subreddit_stats.show(10)

# Time series
daily_stats = aggregator.time_series_aggregation(posts_df, freq="day")
daily_stats.show()

class RedditDataWriter:
    """Write processed Reddit data efficiently."""

    def __init__(self, spark: SparkSession):
        self.spark = spark

    def write_parquet(
        self,
        df,
        path: str,
        partition_by: list = None,
        mode: str = "overwrite"
    ):
        """Write to Parquet (recommended format)."""
        writer = df.write.mode(mode)

        if partition_by:
            writer = writer.partitionBy(*partition_by)

        writer.parquet(path)

    def write_delta(
        self,
        df,
        path: str,
        partition_by: list = None,
        mode: str = "overwrite"
    ):
        """Write to Delta Lake (for ACID transactions)."""
        writer = df.write.format("delta").mode(mode)

        if partition_by:
            writer = writer.partitionBy(*partition_by)

        writer.save(path)

    def optimize_output(self, df, num_files: int = 10):
        """Coalesce for fewer, larger output files."""
        return df.coalesce(num_files)

# Usage
writer = RedditDataWriter(spark)

# Write partitioned by subreddit and date
posts_with_date = posts_df.withColumn(
    "date",
    F.to_date(F.from_unixtime("created_utc"))
)

writer.write_parquet(
    posts_with_date,
    "s3://output/reddit-posts-processed/",
    partition_by=["subreddit", "date"]
)