<jupyter_start><jupyter_text>Subreddit ExplorerThis project was originally completed using 97 million comments from September 2018. However, GitHub has a 100 MB file size limit. Therefore, I created a sample set of approximately 1 million comments and used it below.The results are decent using the sample set (but not as good as the results from the full data set). On the plus side, the entire notebook finishes running in under 2 minutes on my machine.  Initialize Spark- I used a 4-core CPU with 16 GB of memory.- These settings may need to be adjusted for your machine.<jupyter_code>import pyspark as ps
conf = ps.SparkConf().setAll([('spark.executor.memory', '6g'), 
                                   ('spark.executor.cores', '3'),
                                   ('spark.driver.memory','6g')])
spark = ps.sql.SparkSession.builder.config(conf=conf).getOrCreate()<jupyter_output><empty_output><jupyter_text>Load sample comments<jupyter_code>sample_sept = spark.read.csv('sample_sept.csv/', header=True)
sample_sept.count()
sample_sept.printSchema()<jupyter_output>root
 |-- author_fullname: string (nullable = true)
 |-- author: string (nullable = true)
 |-- subreddit: string (nullable = true)
 |-- subreddit_id: string (nullable = true)<jupyter_text>Drop rows where author_fullname = null- These are users where the account was deleted after the comment was created.- The sample comments file should not have any of these.- However, this is an important step if you are using your own source data.<jupyter_code># sample_sept = sample_sept.filter(sample_sept.author_fullname != 'null')<jupyter_output><empty_output><jupyter_text>Convert subreddit_id to base 10- Original format is base 36 (0-9 and a-z)- New column is 'sr_id'<jupyter_code>from pyspark.sql.functions import conv
from pyspark.sql.functions import substring
from pyspark.sql.types import IntegerType
# substring is used to remove prefix 't5_'
base10_df = sample_sept.withColumn('sr_id', 
            conv(substring(sample_sept.subreddit_id, 4, 
            8), 36, 10).cast(IntegerType()))
base10_df.show(5)
base10_df.printSchema()<jupyter_output>root
 |-- author_fullname: string (nullable = true)
 |-- author: string (nullable = true)
 |-- subreddit: string (nullable = true)
 |-- subreddit_id: string (nullable = true)
 |-- sr_id: integer (nullable = true)<jupyter_text>Group rows to create rating column- Each row represents a comment- Multiple comments in a subreddit by the same person need to be grouped- Rating column represents the number of comments (per subreddit)- This is the format required by PySpark ALS<jupyter_code>from pyspark.sql.functions import count
from pyspark.sql.functions import lit
grouped_df = base10_df.groupBy(base10_df.columns).agg(count(lit(1)).alias('rating'))
grouped_df.count()
grouped_df.show(5)<jupyter_output>+---------------+-----------------+---------+------------+-------+------+
|author_fullname|           author|subreddit|subreddit_id|  sr_id|rating|
+---------------+-----------------+---------+------------+-------+------+
|       t2_5uyfq|         tman916x|      nba|    t5_2qo4s|4603564|    61|
|       t2_99q54|         jmeshvrd|     tifu|    t5_2to41|4743505|     2|
|       t2_frjqq|      GhostCheese|pokemongo|    t5_34jka|5250826|    32|
|      t2_103ceu|         Shodan30| Konosuba|    t5_3c02n|5598815|     5|
|    t2_1l3cv8h9|International_Way|AskReddit|    t5_2qh1i|4594374|   162|
+---------------+-----------------+---------+------------+-------+------+
only showing top 5 rows<jupyter_text>Create new author identifier column- The original base 36 author identifier will not work    - Many of them exceed the maximum integer size for PySpark ALS- We will instead create our own author identifier- I tried using monotonically_increasing_id() and ZipWithUniqueId()    - Both created ids larger than the maximum integer size for PySpark ALS    - (When using full data set of 97 million rows)- zipWithIndex() works but uses RDDs which adds additional steps- StringIndexer() is easiest solution that I found<jupyter_code>from pyspark.ml.feature import StringIndexer
from pyspark.sql.functions import col
stringIndexer = StringIndexer(inputCol="author", outputCol="au_id")
si_model = stringIndexer.fit(grouped_df)
auth_df = si_model.transform(grouped_df).withColumn("au_id", 
                                     col("au_id").cast("int"))
# Cache here if you are doing further data exploration.
auth_df.cache()
auth_df.show(5)<jupyter_output>+---------------+-----------------+---------+------------+-------+------+-----+
|author_fullname|           author|subreddit|subreddit_id|  sr_id|rating|au_id|
+---------------+-----------------+---------+------------+-------+------+-----+
|       t2_5uyfq|         tman916x|      nba|    t5_2qo4s|4603564|    61| 6488|
|       t2_99q54|         jmeshvrd|     tifu|    t5_2to41|4743505|     2| 1397|
|       t2_frjqq|      GhostCheese|pokemongo|    t5_34jka|5250826|    32| 1782|
|      t2_103ceu|         Shodan30| Konosuba|    t5_3c02n|5598815|     5| 3508|
|    t2_1l3cv8h9|International_Way|AskReddit|    t5_2qh1i|4594374|   162|  617|
+---------------+-----------------+---------+------------+-------+------+-----+
only showing top 5 rows<jupyter_text>Create dataframe for ALS input- ALS requires (user, item, rating) format- All three fields must be integers<jupyter_code>als_df = auth_df.select(['au_id', 'sr_id', 'rating'])
als_df.printSchema()<jupyter_output>root
 |-- au_id: integer (nullable = true)
 |-- sr_id: integer (nullable = true)
 |-- rating: long (nullable = false)<jupyter_text>--- Run ALS implicit recommendation modelhttps://spark.apache.org/docs/2.2.0/mllib-collaborative-filtering.html<jupyter_code>from pyspark.mllib.recommendation import ALS
als = ALS()
# This setting may help people with less memory
# sc.setCheckpointDir('/tmp')
als_fitted = als.trainImplicit(als_df.rdd, rank=75, iterations=5,
                                  lambda_=0.01, alpha=0.01, seed=10)<jupyter_output><empty_output><jupyter_text>Get items (and users) matrices- PySpark ALS returns these in RDD format- First, we convert these into a Spark dataframe with nested array<jupyter_code>items_rdd = als_fitted.productFeatures()
# users_rdd = als_fitted.userFeatures()
from pyspark.sql.types import Row

def f(x):
    d = {}
    for i in range(len(x)):
        d[str(i)] = x[i]
    return d

items_df = items_rdd.map(lambda x: Row(**f(x))).toDF()
# users_df = users_rdd.map(lambda x: Row(**f(x))).toDF()
items_df.show(5)<jupyter_output>+-------+--------------------+
|      0|                   1|
+-------+--------------------+
|4594300|[0.00937710888683...|
|4594600|[-0.0023872260935...|
|4594700|[-3.4862846951000...|
|4596900|[-3.3788694418035...|
|4598300|[-8.3739380352199...|
+-------+--------------------+
only showing top 5 rows<jupyter_text>Convert into a pandas dataframe as a true matrix- The matrix in the dataframe is nested within a single column- It's much easier to do matrix operations if we convert to a matrix format<jupyter_code>import numpy as np
import pandas as pd

items_pd = pd.DataFrame(np.row_stack(items_df.select('1').collect()))
# users_pd = pd.DataFrame(np.row_stack(users_df.select('1').collect()))
items_pd.head(5)<jupyter_output><empty_output><jupyter_text>Create Sub_list to translate sr_id to row number<jupyter_code>sub_list = pd.DataFrame(items_df.select('0').collect())<jupyter_output><empty_output><jupyter_text>Create table to translate sr_id to subreddit name- We will need this to generate final outputs<jupyter_code>sr_lookup_df = auth_df.select(['subreddit', 'sr_id']).distinct()
sr_lookup = sr_lookup_df.toPandas()<jupyter_output><empty_output><jupyter_text>--- Finally, find subreddits using item-item similarity- This function returns the 10 most similar subreddits to a given input- Based on item-item similarity from the item-feature matrix<jupyter_code>subreddit_set = set(sr_lookup.iloc[:,0])

def item_item_sim(search_name):
    if str(search_name) not in subreddit_set:
        return "Error: " + str(search_name) + " is not an active subreddit"
    
    # Get subreddit id number and translate to row number
    search_id = sr_lookup[sr_lookup['subreddit'] == str(search_name)].iloc[0,1]
    search_row_num = sub_list[sub_list[0] == search_id].index[0]
    
    # Get item-feature row and calculate dot product
    item_feature_np = np.asarray(items_pd.iloc[search_row_num, :])
    item_dp = np.dot(item_feature_np, items_pd.T)
    
    # Get top ten similar rows, translate to subreddit name
    sim_rows = list(item_dp.argsort())[-11:-1][::-1]
    out_ids = sub_list.iloc[sim_rows]
    out_names = list(out_ids.join(sr_lookup.set_index('sr_id'), on=0).loc[:,'subreddit'])
    
    return out_names
item_item_sim('nfl')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16220526255837794, 'fraction_numerical': 0.04271557124957284, 'mean_word_length': 3.268351968886728, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 47, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '186582', 'license': 'MIT', 'n_tokens_mistral': 3090, 'n_tokens_neox': 2867, 'n_words': 730, 'path': 'sub_explorer.ipynb', 'repo_name': 'pjhur/subreddit_explorer', 'size': '8779'}