old/term_cosine_similarity.py

   1 from pyspark.sql import functions as f
   2 from pyspark.sql import SparkSession
   3 from pyspark.sql import Window
   4 from pyspark.mllib.linalg.distributed import RowMatrix, CoordinateMatrix
   5 import numpy as np
   6 import pyarrow
   7 import pandas as pd
   8 import fire
   9 from itertools import islice
  10 from pathlib import Path
  11 from similarities_helper import prep_tfidf_entries, read_tfidf_matrix, column_similarities, select_topN
  12 import scipy
  13
  14 # outfile='test_similarities_500.feather';
  15 # min_df = None;
  16 # included_subreddits=None; topN=100; exclude_phrases=True;
  17 def term_cosine_similarities(outfile, min_df = None, included_subreddits=None, topN=500, exclude_phrases=False):
  18     spark = SparkSession.builder.getOrCreate()
  19     conf = spark.sparkContext.getConf()
  20     print(outfile)
  21     print(exclude_phrases)
  22
  23     tfidf = spark.read.parquet('/gscratch/comdata/output/reddit_similarity/tfidf/subreddit_terms.parquet')
  24
  25     if included_subreddits is None:
  26         included_subreddits = select_topN_subreddits(topN)
  27     else:
  28         included_subreddits = set(open(included_subreddits))
  29
  30     if exclude_phrases == True:
  31         tfidf = tfidf.filter(~f.col(term).contains("_"))
  32
  33     print("creating temporary parquet with matrix indicies")
  34     tempdir = prep_tfidf_entries(tfidf, 'term', min_df, included_subreddits)
  35     tfidf = spark.read.parquet(tempdir.name)
  36     subreddit_names = tfidf.select(['subreddit','subreddit_id_new']).distinct().toPandas()
  37     subreddit_names = subreddit_names.sort_values("subreddit_id_new")
  38     subreddit_names['subreddit_id_new'] = subreddit_names['subreddit_id_new'] - 1
  39     spark.stop()
  40
  41     print("loading matrix")
  42     mat = read_tfidf_matrix(tempdir.name,'term')
  43     print('computing similarities')
  44     sims = column_similarities(mat)
  45     del mat
  46
  47     sims = pd.DataFrame(sims.todense())
  48     sims = sims.rename({i:sr for i, sr in enumerate(subreddit_names.subreddit.values)},axis=1)
  49     sims['subreddit'] = subreddit_names.subreddit.values
  50
  51     p = Path(outfile)
  52
  53     output_feather =  Path(str(p).replace("".join(p.suffixes), ".feather"))
  54     output_csv =  Path(str(p).replace("".join(p.suffixes), ".csv"))
  55     output_parquet =  Path(str(p).replace("".join(p.suffixes), ".parquet"))
  56
  57     sims.to_feather(outfile)
  58     tempdir.cleanup()
  59
  60 if __name__ == '__main__':
  61     fire.Fire(term_cosine_similarities)