From: Nate E TeBlunthuis Date: Thu, 28 Jan 2021 04:09:23 +0000 (-0800) Subject: Merge branch 'master' of code:cdsc_reddit X-Git-Url: https://code.communitydata.science/cdsc_reddit.git/commitdiff_plain/b4dd9acbd8e499d87714413b8260240341ebf7d7?hp=dbe4c87f8b70e348705319bdf3d0a10b64d763b5 Merge branch 'master' of code:cdsc_reddit --- diff --git a/clustering/Makefile b/clustering/Makefile new file mode 100644 index 0000000..115b218 --- /dev/null +++ b/clustering/Makefile @@ -0,0 +1,10 @@ +#srun_cdsc='srun -p comdata-int -A comdata --time=300:00:00 --time-min=00:15:00 --mem=100G --ntasks=1 --cpus-per-task=28' +all:/gscratch/comdata/output/reddit_clustering/comment_authors_10000.feather /gscratch/comdata/output/reddit_clustering/comment_terms_10000.feather + +/gscratch/comdata/output/reddit_clustering/comment_authors_10000.feather:clustering.py /gscratch/comdata/output/reddit_similarity/comment_authors_10000.feather +# $srun_cdsc python3 + ./clustering.py /gscratch/comdata/output/reddit_similarity/comment_authors_10000.feather /gscratch/comdata/output/reddit_clustering/comment_authors_10000.feather ---max_iter=400 --convergence_iter=15 --preference_quantile=0.85 --damping=0.85 + +/gscratch/comdata/output/reddit_clustering/comment_terms_10000.feather:clustering.py /gscratch/comdata/output/reddit_similarity/comment_terms_10000.feather +# $srun_cdsc python3 + ./clustering.py /gscratch/comdata/output/reddit_similarity/comment_terms_10000.feather /gscratch/comdata/output/reddit_clustering/comment_terms_10000.feather ---max_iter=1000 --convergence_iter=15 --preference_quantile=0.9 --damping=0.5 diff --git a/clustering/affinity/subreddit_comment_authors_10000_a.feather b/clustering/affinity/subreddit_comment_authors_10000_a.feather new file mode 100644 index 0000000..21e15e4 Binary files /dev/null and b/clustering/affinity/subreddit_comment_authors_10000_a.feather differ diff --git a/clustering/clustering.py b/clustering/clustering.py old mode 100644 new mode 100755 index 38af31c..e652304 --- a/clustering/clustering.py +++ b/clustering/clustering.py @@ -1,12 +1,15 @@ +#!/usr/bin/env python3 + import pandas as pd import numpy as np from sklearn.cluster import AffinityPropagation import fire -def affinity_clustering(similarities, output, damping=0.5, max_iter=100000, convergence_iter=30, preference_quantile=0.5, random_state=1968): +def affinity_clustering(similarities, output, damping=0.9, max_iter=100000, convergence_iter=30, preference_quantile=0.5, random_state=1968, verbose=True): ''' similarities: feather file with a dataframe of similarity scores preference_quantile: parameter controlling how many clusters to make. higher values = more clusters. 0.85 is a good value with 3000 subreddits. + damping: parameter controlling how iterations are merged. Higher values make convergence faster and more dependable. 0.85 is a good value for the 10000 subreddits by author. ''' df = pd.read_feather(similarities) @@ -16,6 +19,8 @@ def affinity_clustering(similarities, output, damping=0.5, max_iter=100000, conv preference = np.quantile(mat,preference_quantile) + print(f"preference is {preference}") + print("data loaded") clustering = AffinityPropagation(damping=damping, @@ -24,6 +29,7 @@ def affinity_clustering(similarities, output, damping=0.5, max_iter=100000, conv copy=False, preference=preference, affinity='precomputed', + verbose=verbose, random_state=random_state).fit(mat) diff --git a/density/Makefile b/density/Makefile new file mode 100644 index 0000000..43075a4 --- /dev/null +++ b/density/Makefile @@ -0,0 +1,7 @@ +all: /gscratch/comdata/output/reddit_density/comment_terms_10000.feather /gscratch/comdata/output/reddit_density/comment_authors_10000.feather + +/gscratch/comdata/output/reddit_density/comment_terms_10000.feather:overlap_density.py /gscratch/comdata/output/reddit_similarity/comment_terms_10000.feather /gscratch/comdata/output/reddit_similarity/comment_terms_10000.feather + python3 overlap_density.py terms --inpath="/gscratch/comdata/output/reddit_similarity/comment_terms_10000.feather" --outpath="/gscratch/comdata/output/reddit_density/comment_terms_10000.feather" --agg=pd.DataFrame.sum + +/gscratch/comdata/output/reddit_density/comment_authors_10000.feather:overlap_density.py /gscratch/comdata/output/reddit_similarity/comment_authors_10000.feather + python3 overlap_density.py authors --inpath="/gscratch/comdata/output/reddit_similarity/comment_authors_10000.feather" --outpath="/gscratch/comdata/output/reddit_density/comment_authors_10000.feather" --agg=pd.DataFrame.sum diff --git a/density/job_script.sh b/density/job_script.sh new file mode 100755 index 0000000..553d0a1 --- /dev/null +++ b/density/job_script.sh @@ -0,0 +1,4 @@ +#!/usr/bin/bash +start_spark_cluster.sh +spark-submit --master spark://$(hostname):18899 overlap_density.py wang_overlaps --inpath=/gscratch/comdata/output/reddit_similarity/tfidf_weekly/comment_authors.parquet --to_date=2020-04-13 +stop-all.sh diff --git a/density/overlap_density.py b/density/overlap_density.py new file mode 100644 index 0000000..a1e9f6d --- /dev/null +++ b/density/overlap_density.py @@ -0,0 +1,76 @@ +import pandas as pd +from pandas.core.groupby import DataFrameGroupBy as GroupBy +import fire +import numpy as np +import sys +sys.path.append("..") +sys.path.append("../similarities") +from similarities.similarities_helper import read_tfidf_matrix, reindex_tfidf, reindex_tfidf_time_interval + +# this is the mean of the ratio of the overlap to the focal size. +# mean shared membership per focal community member +# the input is the author tf-idf matrix + +def overlap_density(inpath, outpath, agg = pd.DataFrame.sum): + df = pd.read_feather(inpath) + df = df.drop('subreddit',1) + np.fill_diagonal(df.values,0) + df = agg(df, 0).reset_index() + df = df.rename({0:'overlap_density'},axis='columns') + df.to_feather(outpath) + return df + +def overlap_density_weekly(inpath, outpath, agg = GroupBy.sum): + df = pd.read_parquet(inpath) + # exclude the diagonal + df = df.loc[df.subreddit != df.variable] + res = agg(df.groupby(['subreddit','week'])).reset_index() + res.to_feather(outpath) + return res + + +# inpath="/gscratch/comdata/output/reddit_similarity/tfidf/comment_authors.parquet"; +# min_df=1; +# included_subreddits=None; +# topN=10000; +# outpath="/gscratch/comdata/output/reddit_density/wang_overlaps_10000.feather" + +# to_date=2019-10-28 + + +def author_overlap_density(inpath="/gscratch/comdata/output/reddit_similarity/comment_authors_10000.feather", + outpath="/gscratch/comdata/output/reddit_density/comment_authors_10000.feather", agg=pd.DataFrame.sum): + if type(agg) == str: + agg = eval(agg) + + overlap_density(inpath, outpath, agg) + +def term_overlap_density(inpath="/gscratch/comdata/output/reddit_similarity/comment_terms_10000.feather", + outpath="/gscratch/comdata/output/reddit_density/comment_term_similarity_10000.feather", agg=pd.DataFrame.sum): + + if type(agg) == str: + agg = eval(agg) + + overlap_density(inpath, outpath, agg) + +def author_overlap_density_weekly(inpath="/gscratch/comdata/output/reddit_similarity/subreddit_authors_10000_weekly.parquet", + outpath="/gscratch/comdata/output/reddit_density/comment_authors_10000_weekly.feather", agg=GroupBy.sum): + if type(agg) == str: + agg = eval(agg) + + overlap_density_weekly(inpath, outpath, agg) + +def term_overlap_density_weekly(inpath="/gscratch/comdata/output/reddit_similarity/comment_terms_10000_weekly.parquet", + outpath="/gscratch/comdata/output/reddit_density/comment_terms_10000_weekly.parquet", agg=GroupBy.sum): + if type(agg) == str: + agg = eval(agg) + + overlap_density_weekly(inpath, outpath, agg) + + +if __name__ == "__main__": + fire.Fire({'authors':author_overlap_density, + 'terms':term_overlap_density, + 'author_weekly':author_overlap_density_weekly, + 'term_weekly':term_overlap_density_weekly, + 'wang_overlaps':wang_overlap_density}) diff --git a/old/.#tfidf_authors.py b/old/.#tfidf_authors.py deleted file mode 120000 index 8972575..0000000 --- a/old/.#tfidf_authors.py +++ /dev/null @@ -1 +0,0 @@ -nathante@n2347.hyak.local.31061:1602221800 \ No newline at end of file diff --git a/old/.#tfidf_comments_weekly.py b/old/.#tfidf_comments_weekly.py deleted file mode 120000 index 8972575..0000000 --- a/old/.#tfidf_comments_weekly.py +++ /dev/null @@ -1 +0,0 @@ -nathante@n2347.hyak.local.31061:1602221800 \ No newline at end of file diff --git a/similarities/#cosine_similarities.py# b/similarities/#cosine_similarities.py# deleted file mode 100644 index ae080d5..0000000 --- a/similarities/#cosine_similarities.py# +++ /dev/null @@ -1,73 +0,0 @@ -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 - - -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) - - tfidf = spark.read.parquet(infile) - - if included_subreddits is None: - included_subreddits = select_topN_subreddits(topN) - else: - included_subreddits = set(open(included_subreddits)) - - 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', - 'term', - outfile, - min_df, - included_subreddits, - topN, - exclude_phrases) - -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', - 'author', - outfile, - min_df, - included_subreddits, - topN, - exclude_phrases=False) - -if __name__ == "__main__": - fire.Fire({'term':term_cosine_similarities, - 'author':author_cosine_similarities}) - diff --git a/similarities/.#cosine_similarities.py b/similarities/.#cosine_similarities.py deleted file mode 120000 index 8972575..0000000 --- a/similarities/.#cosine_similarities.py +++ /dev/null @@ -1 +0,0 @@ -nathante@n2347.hyak.local.31061:1602221800 \ No newline at end of file diff --git a/similarities/.#tfidf_weekly.py b/similarities/.#tfidf_weekly.py deleted file mode 120000 index 8972575..0000000 --- a/similarities/.#tfidf_weekly.py +++ /dev/null @@ -1 +0,0 @@ -nathante@n2347.hyak.local.31061:1602221800 \ No newline at end of file diff --git a/similarities/Makefile b/similarities/Makefile index 89a908f..51fd0fa 100644 --- a/similarities/Makefile +++ b/similarities/Makefile @@ -1,2 +1,13 @@ +all: /gscratch/comdata/output/reddit_similarity/subreddit_comment_terms_25000.parquet /gscratch/comdata/output/reddit_similarity/subreddit_comment_authors_25000.parquet /gscratch/comdata/output/reddit_similarity/subreddit_comment_authors_10000.parquet /gscratch/comdata/output/reddit_similarity/comment_terms_10000_weekly.parquet + +/gscratch/comdata/output/reddit_similarity/subreddit_comment_terms_25000.parquet: cosine_similarities.py /gscratch/comdata/output/reddit_similarity/tfidf/comment_terms.parquet + start_spark_and_run.sh 1 cosine_similarities.py term --outfile=/gscratch/comdata/output/reddit_similarity/subreddit_comment_terms_25000.feather + +/gscratch/comdata/output/reddit_similarity/subreddit_comment_authors_25000.parquet: cosine_similarities.py /gscratch/comdata/output/reddit_similarity/tfidf/comment_authors.parquet + start_spark_and_run.sh 1 cosine_similarities.py author --outfile=/gscratch/comdata/output/reddit_similarity/subreddit_comment_authors_25000.feather + /gscratch/comdata/output/reddit_similarity/subreddit_comment_authors_10000.parquet: cosine_similarities.py /gscratch/comdata/output/reddit_similarity/tfidf/comment_authors.parquet - start_spark_and_run.sh 1 cosine_similarities.py author --outfile=/gscratch/comdata/output/reddit_similarity/subreddit_comment_authors_10000.parquet + start_spark_and_run.sh 1 cosine_similarities.py author --outfile=/gscratch/comdata/output/reddit_similarity/subreddit_comment_authors_10000.feather + +/gscratch/comdata/output/reddit_similarity/comment_terms_10000_weekly.parquet: cosine_similarities.py /gscratch/comdata/output/reddit_similarity/tfidf/comment_authors.parquet + start_spark_and_run.sh 1 weekly_cosine_similarities.py term --outfile=/gscratch/comdata/output/reddit_similarity/subreddit_comment_terms_10000_weely.parquet diff --git a/similarities/cosine_similarities.py b/similarities/cosine_similarities.py index ae080d5..609e477 100644 --- a/similarities/cosine_similarities.py +++ b/similarities/cosine_similarities.py @@ -1,64 +1,21 @@ -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 +def cosine_similarities(infile, term_colname, outfile, min_df=None, included_subreddits=None, topN=500, exclude_phrases=False,from_date=None, to_date=None): + return similiarities(infile=infile, simfunc=column_similarities, term_colname=term_colname, outfile=outfile, min_df=min_df, included_subreddits=included_subreddits, topN=topN, exclude_phrases=exclude_phrases,from_date=from_date, to_date=to_date) -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) - - tfidf = spark.read.parquet(infile) - - if included_subreddits is None: - included_subreddits = select_topN_subreddits(topN) - else: - included_subreddits = set(open(included_subreddits)) - - 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): +def term_cosine_similarities(outfile, min_df=None, included_subreddits=None, topN=500, exclude_phrases=False, from_date=None, to_date=None): return cosine_similarities('/gscratch/comdata/output/reddit_similarity/tfidf/comment_terms.parquet', 'term', outfile, min_df, included_subreddits, topN, - exclude_phrases) + exclude_phrasesby.) -def author_cosine_similarities(outfile, min_df=2, included_subreddits=None, topN=10000): +def author_cosine_similarities(outfile, min_df=2, included_subreddits=None, topN=10000, from_date=None, to_date=None): return cosine_similarities('/gscratch/comdata/output/reddit_similarity/tfidf/comment_authors.parquet', 'author', outfile, diff --git a/similarities/job_script.sh b/similarities/job_script.sh index 1c2cacf..e79a061 100755 --- a/similarities/job_script.sh +++ b/similarities/job_script.sh @@ -1,4 +1,4 @@ #!/usr/bin/bash start_spark_cluster.sh -spark-submit --master spark://$(hostname):18899 cosine_similarities.py author --outfile=/gscratch/comdata/output/reddit_similarity/subreddit_comment_authors_10000.parquet +spark-submit --master spark://$(hostname):18899 wang_similarity.py --infile=/gscratch/comdata/output/reddit_similarity/tfidf/comment_authors.parquet --max_df=10 --outfile=/gscratch/comdata/output/reddit_similarity/wang_similarity_1000_max10.feather stop-all.sh diff --git a/similarities/similarities_helper.py b/similarities/similarities_helper.py index 88c830c..69516a6 100644 --- a/similarities/similarities_helper.py +++ b/similarities/similarities_helper.py @@ -1,3 +1,4 @@ +from pyspark.sql import SparkSession from pyspark.sql import Window from pyspark.sql import functions as f from enum import Enum @@ -5,15 +6,108 @@ from pyspark.mllib.linalg.distributed import CoordinateMatrix from tempfile import TemporaryDirectory import pyarrow import pyarrow.dataset as ds -from scipy.sparse import csr_matrix +from scipy.sparse import csr_matrix, issparse import pandas as pd import numpy as np import pathlib +from datetime import datetime +from pathlib import Path class tf_weight(Enum): MaxTF = 1 Norm05 = 2 +infile = "/gscratch/comdata/output/reddit_similarity/tfidf_weekly/comment_authors.parquet" + +def reindex_tfidf_time_interval(infile, term_colname, min_df=None, max_df=None, included_subreddits=None, topN=500, exclude_phrases=False, from_date=None, to_date=None): + term = term_colname + term_id = term + '_id' + term_id_new = term + '_id_new' + + spark = SparkSession.builder.getOrCreate() + conf = spark.sparkContext.getConf() + print(exclude_phrases) + tfidf_weekly = spark.read.parquet(infile) + + # create the time interval + if from_date is not None: + if type(from_date) is str: + from_date = datetime.fromisoformat(from_date) + + tfidf_weekly = tfidf_weekly.filter(tfidf_weekly.week >= from_date) + + if to_date is not None: + if type(to_date) is str: + to_date = datetime.fromisoformat(to_date) + tfidf_weekly = tfidf_weekly.filter(tfidf_weekly.week < to_date) + + tfidf = tfidf_weekly.groupBy(["subreddit","week", term_id, term]).agg(f.sum("tf").alias("tf")) + tfidf = _calc_tfidf(tfidf, term_colname, tf_weight.Norm05) + tempdir = prep_tfidf_entries(tfidf, term_colname, min_df, max_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 + return(tempdir, subreddit_names) + +def reindex_tfidf(infile, term_colname, min_df=None, max_df=None, included_subreddits=None, topN=500, exclude_phrases=False): + spark = SparkSession.builder.getOrCreate() + conf = spark.sparkContext.getConf() + print(exclude_phrases) + + tfidf = spark.read.parquet(infile) + + if included_subreddits is None: + included_subreddits = select_topN_subreddits(topN) + else: + included_subreddits = set(open(included_subreddits)) + + 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, max_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() + return (tempdir, subreddit_names) + +def similarities(infile, simfunc, term_colname, outfile, min_df=None, max_df=None, included_subreddits=None, topN=500, exclude_phrases=False, from_date=None, to_date=None): + + if from_date is not None or to_date is not None: + tempdir, subreddit_names = reindex_tfidf_time_interval(infile, term_colname='author', min_df=min_df, max_df=max_df, included_subreddits=included_subreddits, topN=topN, exclude_phrases=False, from_date=from_date, to_date=to_date) + + else: + tempdir, subreddit_names = reindex_tfidf(infile, term_colname='author', min_df=min_df, max_df=max_df, included_subreddits=included_subreddits, topN=topN, exclude_phrases=False) + + print("loading matrix") + # mat = read_tfidf_matrix("term_tfidf_entries7ejhvnvl.parquet", term_colname) + mat = read_tfidf_matrix(tempdir.name, term_colname) + print('computing similarities') + sims = simfunc(mat) + del mat + + if issparse(sims): + sims = sims.todense() + + print(f"shape of sims:{sims.shape}") + print(f"len(subreddit_names.subreddit.values):{len(subreddit_names.subreddit.values)}") + sims = pd.DataFrame(sims) + 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 read_tfidf_matrix_weekly(path, term_colname, week): term = term_colname term_id = term + '_id' @@ -33,8 +127,6 @@ def write_weekly_similarities(path, sims, week, names): sims = sims.melt(id_vars=['subreddit','week'],value_vars=names.subreddit.values) sims.to_parquet(p / week.isoformat()) - - def read_tfidf_matrix(path,term_colname): term = term_colname term_id = term + '_id' @@ -44,6 +136,15 @@ def read_tfidf_matrix(path,term_colname): entries = dataset.to_table(columns=['tf_idf','subreddit_id_new',term_id_new]).to_pandas() return(csr_matrix((entries.tf_idf,(entries[term_id_new]-1, entries.subreddit_id_new-1)))) +def column_overlaps(mat): + non_zeros = (mat != 0).astype('double') + + intersection = non_zeros.T @ non_zeros + card1 = non_zeros.sum(axis=0) + den = np.add.outer(card1,card1) - intersection + + return intersection / den + def column_similarities(mat): norm = np.matrix(np.power(mat.power(2).sum(axis=0),0.5,dtype=np.float32)) mat = mat.multiply(1/norm) @@ -51,13 +152,16 @@ def column_similarities(mat): return(sims) -def prep_tfidf_entries_weekly(tfidf, term_colname, min_df, included_subreddits): +def prep_tfidf_entries_weekly(tfidf, term_colname, min_df, max_df, included_subreddits): term = term_colname term_id = term + '_id' term_id_new = term + '_id_new' if min_df is None: min_df = 0.1 * len(included_subreddits) + tfidf = tfidf.filter(f.col('count') >= min_df) + if max_df is not None: + tfidf = tfidf.filter(f.col('count') <= max_df) tfidf = tfidf.filter(f.col("subreddit").isin(included_subreddits)) @@ -86,19 +190,22 @@ def prep_tfidf_entries_weekly(tfidf, term_colname, min_df, included_subreddits): return(tempdir) -def prep_tfidf_entries(tfidf, term_colname, min_df, included_subreddits): +def prep_tfidf_entries(tfidf, term_colname, min_df, max_df, included_subreddits): term = term_colname term_id = term + '_id' term_id_new = term + '_id_new' if min_df is None: min_df = 0.1 * len(included_subreddits) + tfidf = tfidf.filter(f.col('count') >= min_df) + if max_df is not None: + tfidf = tfidf.filter(f.col('count') <= max_df) tfidf = tfidf.filter(f.col("subreddit").isin(included_subreddits)) # reset the subreddit ids sub_ids = tfidf.select('subreddit_id').distinct() - sub_ids = sub_ids.withColumn("subreddit_id_new",f.row_number().over(Window.orderBy("subreddit_id"))) + sub_ids = sub_ids.withColumn("subreddit_id_new", f.row_number().over(Window.orderBy("subreddit_id"))) tfidf = tfidf.join(sub_ids,'subreddit_id') # only use terms in at least min_df included subreddits @@ -221,15 +328,9 @@ def build_weekly_tfidf_dataset(df, include_subs, term_colname, tf_family=tf_weig return df - - -def build_tfidf_dataset(df, include_subs, term_colname, tf_family=tf_weight.Norm05): - +def _calc_tfidf(df, term_colname, tf_family): term = term_colname term_id = term + '_id' - # aggregate counts by week. now subreddit-term is distinct - df = df.filter(df.subreddit.isin(include_subs)) - df = df.groupBy(['subreddit',term]).agg(f.sum('tf').alias('tf')) max_subreddit_terms = df.groupby(['subreddit']).max('tf') # subreddits are unique max_subreddit_terms = max_subreddit_terms.withColumnRenamed('max(tf)','sr_max_tf') @@ -240,9 +341,7 @@ def build_tfidf_dataset(df, include_subs, term_colname, tf_family=tf_weight.Norm # group by term. term is unique idf = df.groupby([term]).count() - N_docs = df.select('subreddit').distinct().count() - # add a little smoothing to the idf idf = idf.withColumn('idf',f.log(N_docs/(1+f.col('count')))+1) @@ -271,6 +370,18 @@ def build_tfidf_dataset(df, include_subs, term_colname, tf_family=tf_weight.Norm df = df.withColumn("tf_idf", (0.5 + 0.5 * df.relative_tf) * df.idf) return df + + +def build_tfidf_dataset(df, include_subs, term_colname, tf_family=tf_weight.Norm05): + term = term_colname + term_id = term + '_id' + # aggregate counts by week. now subreddit-term is distinct + df = df.filter(df.subreddit.isin(include_subs)) + df = df.groupBy(['subreddit',term]).agg(f.sum('tf').alias('tf')) + + df = _calc_tfidf(df, term_colname, tf_family) + + return df def select_topN_subreddits(topN, path="/gscratch/comdata/output/reddit_similarity/subreddits_by_num_comments.csv"): rankdf = pd.read_csv(path) diff --git a/similarities/tfidf.py b/similarities/tfidf.py index 5b1c0c9..b7b4e63 100644 --- a/similarities/tfidf.py +++ b/similarities/tfidf.py @@ -45,7 +45,7 @@ def tfidf_terms(outpath='/gscratch/comdata/output/reddit_similarity/tfidf/commen [] ) -def tfidf_authors_weekly(outpath='/gscratch/comdata/output/reddit_similarity/tfidf/comment_authors.parquet', +def tfidf_authors_weekly(outpath='/gscratch/comdata/output/reddit_similarity/tfidf_weekly/comment_authors.parquet', topN=25000): return tfidf_weekly("/gscratch/comdata/output/reddit_ngrams/comment_authors.parquet", @@ -55,7 +55,7 @@ def tfidf_authors_weekly(outpath='/gscratch/comdata/output/reddit_similarity/tfi ['[deleted]','AutoModerator'] ) -def tfidf_terms_weekly(outpath='/gscratch/comdata/output/reddit_similarity/tfidf/comment_terms.parquet', +def tfidf_terms_weekly(outpath='/gscratch/comdata/output/reddit_similarity/tfidf_weekly/comment_terms.parquet', topN=25000): return tfidf_weekly("/gscratch/comdata/output/reddit_ngrams/comment_terms.parquet", diff --git a/similarities/top_subreddits_by_comments.py b/similarities/top_subreddits_by_comments.py index 214c7e0..1197b51 100644 --- a/similarities/top_subreddits_by_comments.py +++ b/similarities/top_subreddits_by_comments.py @@ -5,6 +5,10 @@ from pyspark.sql import Window spark = SparkSession.builder.getOrCreate() conf = spark.sparkContext.getConf() +submissions = spark.read.parquet("/gscratch/comdata/output/reddit_submissions_by_subreddit.parquet") + +prop_nsfw = submissions.select(['subreddit','over_18']).groupby('subreddit').agg(f.mean(f.col('over_18').astype('double')).alias('prop_nsfw')) + df = spark.read.parquet("/gscratch/comdata/output/reddit_comments_by_subreddit.parquet") # remove /u/ pages @@ -12,6 +16,9 @@ df = df.filter(~df.subreddit.like("u_%")) df = df.groupBy('subreddit').agg(f.count('id').alias("n_comments")) +df = df.join(prop_nsfw,on='subreddit') +df = df.filter(df.prop_nsfw < 0.5) + win = Window.orderBy(f.col('n_comments').desc()) df = df.withColumn('comments_rank', f.rank().over(win)) diff --git a/similarities/wang_similarity.py b/similarities/wang_similarity.py new file mode 100644 index 0000000..99dc3cb --- /dev/null +++ b/similarities/wang_similarity.py @@ -0,0 +1,18 @@ +from similarities_helper import similarities +import numpy as np +import fire + +def wang_similarity(mat): + non_zeros = (mat != 0).astype(np.float32) + intersection = non_zeros.T @ non_zeros + return intersection + + +infile="/gscratch/comdata/output/reddit_similarity/tfidf/comment_authors.parquet"; outfile="/gscratch/comdata/output/reddit_similarity/wang_similarity_10000.feather"; min_df=1; included_subreddits=None; topN=10000; exclude_phrases=False; from_date=None; to_date=None + +def wang_overlaps(infile, outfile="/gscratch/comdata/output/reddit_similarity/wang_similarity_10000.feather", min_df=1, max_df=None, included_subreddits=None, topN=10000, exclude_phrases=False, from_date=None, to_date=None): + + return similarities(infile=infile, simfunc=wang_similarity, term_colname='author', outfile=outfile, min_df=min_df, max_df=None, included_subreddits=included_subreddits, topN=topN, exclude_phrases=exclude_phrases, from_date=from_date, to_date=to_date) + +if __name__ == "__main__": + fire.Fire(wang_overlaps) diff --git a/similarities/weekly_cosine_similarities.py b/similarities/weekly_cosine_similarities.py index 2b3c90b..4d496f0 100644 --- a/similarities/weekly_cosine_similarities.py +++ b/similarities/weekly_cosine_similarities.py @@ -35,7 +35,7 @@ def cosine_similarities_weekly(tfidf_path, outfile, term_colname, min_df = None, subreddit_names['subreddit_id_new'] = subreddit_names['subreddit_id_new'] - 1 spark.stop() - weeks = list(subreddit_names.week.drop_duplicates()) + weeks = sorted(list(subreddit_names.week.drop_duplicates())) for week in weeks: print(f"loading matrix: {week}") mat = read_tfidf_matrix_weekly(tempdir.name, term_colname, week) diff --git a/visualization/Makefile b/visualization/Makefile new file mode 100644 index 0000000..e69de29