update pushshift dumps.

[cdsc_reddit.git] / similarities / cosine_similarities.py

diff --git a/similarities/cosine_similarities.py b/similarities/cosine_similarities.py
index ae080d5d6b614dac46de8a45ce5237cd5d94bd53..b9bab17b44a97a78070bd17f7cfc1dad407d0130 100644 (file)
--- a/similarities/cosine_similarities.py
+++ b/similarities/cosine_similarities.py
@@ -1,73 +1,60 @@
-from pyspark.sql import functions as f
-from pyspark.sql import SparkSession
 import pandas as pd
 import fire
 from pathlib import Path
-from similarities_helper import prep_tfidf_entries, read_tfidf_matrix, select_topN_subreddits
+from similarities_helper import similarities, column_similarities
+from functools import partial
 
+def cosine_similarities(infile, term_colname, outfile, min_df=None, max_df=None, included_subreddits=None, topN=500, exclude_phrases=False, from_date=None, to_date=None, tfidf_colname='tf_idf'):
 
-def cosine_similarities(infile, term_colname, outfile, min_df=None, included_subreddits=None, topN=500, exclude_phrases=False):
-    spark = SparkSession.builder.getOrCreate()
-    conf = spark.sparkContext.getConf()
-    print(outfile)
-    print(exclude_phrases)
+    return similarities(infile=infile, simfunc=column_similarities, term_colname=term_colname, outfile=outfile, min_df=min_df, max_df=max_df, included_subreddits=included_subreddits, topN=topN, exclude_phrases=exclude_phrases,from_date=from_date, to_date=to_date, tfidf_colname=tfidf_colname)
 
-    tfidf = spark.read.parquet(infile)
+# change so that these take in an input as an optional argument (for speed, but also for idf).
+def term_cosine_similarities(outfile, min_df=None, max_df=None, included_subreddits=None, topN=500, exclude_phrases=False, from_date=None, to_date=None):
 
-    if included_subreddits is None:
-        included_subreddits = select_topN_subreddits(topN)
-    else:
-        included_subreddits = set(open(included_subreddits))
+def term_cosine_similarities(outfile, infile='/gscratch/comdata/output/reddit_similarity/tfidf/comment_terms_100k.parquet', min_df=None, max_df=None, included_subreddits=None, topN=500, exclude_phrases=False, from_date=None, to_date=None):
 
-    if exclude_phrases == True:
-        tfidf = tfidf.filter(~f.col(term_colname).contains("_"))
-
-    print("creating temporary parquet with matrix indicies")
-    tempdir = prep_tfidf_entries(tfidf, term_colname, min_df, included_subreddits)
-    tfidf = spark.read.parquet(tempdir.name)
-    subreddit_names = tfidf.select(['subreddit','subreddit_id_new']).distinct().toPandas()
-    subreddit_names = subreddit_names.sort_values("subreddit_id_new")
-    subreddit_names['subreddit_id_new'] = subreddit_names['subreddit_id_new'] - 1
-    spark.stop()
-
-    print("loading matrix")
-    mat = read_tfidf_matrix(tempdir.name, term_colname)
-    print('computing similarities')
-    sims = column_similarities(mat)
-    del mat
-    
-    sims = pd.DataFrame(sims.todense())
-    sims = sims.rename({i:sr for i, sr in enumerate(subreddit_names.subreddit.values)}, axis=1)
-    sims['subreddit'] = subreddit_names.subreddit.values
-
-    p = Path(outfile)
-
-    output_feather =  Path(str(p).replace("".join(p.suffixes), ".feather"))
-    output_csv =  Path(str(p).replace("".join(p.suffixes), ".csv"))
-    output_parquet =  Path(str(p).replace("".join(p.suffixes), ".parquet"))
-
-    sims.to_feather(outfile)
-    tempdir.cleanup()
-
-def term_cosine_similarities(outfile, min_df=None, included_subreddits=None, topN=500, exclude_phrases=False):
-    return cosine_similarities('/gscratch/comdata/output/reddit_similarity/tfidf/comment_terms.parquet',
+    return cosine_similarities(infile,
                                'term',
                                outfile,
                                min_df,
+                               max_df,
                                included_subreddits,
                                topN,
-                               exclude_phrases)
+                               exclude_phrases,
+                               from_date,
+                               to_date
+                               )
 
-def author_cosine_similarities(outfile, min_df=2, included_subreddits=None, topN=10000):
-    return cosine_similarities('/gscratch/comdata/output/reddit_similarity/tfidf/comment_authors.parquet',
+def author_cosine_similarities(outfile, infile='/gscratch/comdata/output/reddit_similarity/tfidf/comment_authors_100k.parquet', min_df=2, max_df=None, included_subreddits=None, topN=10000, from_date=None, to_date=None):
+    return cosine_similarities(infile,
                                'author',
                                outfile,
                                min_df,
+                               max_df,
                                included_subreddits,
                                topN,
-                               exclude_phrases=False)
+                               exclude_phrases=False,
+                               from_date=from_date,
+                               to_date=to_date
+                               )
+
+def author_tf_similarities(outfile, infile='/gscratch/comdata/output/reddit_similarity/tfidf/comment_authors_100k.parquet', min_df=2, max_df=None, included_subreddits=None, topN=10000, from_date=None, to_date=None):
+    return cosine_similarities(infile,
+                               'author',
+                               outfile,
+                               min_df,
+                               max_df,
+                               included_subreddits,
+                               topN,
+                               exclude_phrases=False,
+                               from_date=from_date,
+                               to_date=to_date,
+                               tfidf_colname='relative_tf'
+                               )
+
 
 if __name__ == "__main__":
     fire.Fire({'term':term_cosine_similarities,
-               'author':author_cosine_similarities})
+               'author':author_cosine_similarities,
+               'author-tf':author_tf_similarities})