From 2740f55915d6ecca7c5cd800747d9687c4cd9245 Mon Sep 17 00:00:00 2001 From: Nate E TeBlunthuis Date: Sun, 23 Aug 2020 11:57:55 -0700 Subject: [PATCH] Compute IDF for terms and authors. --- checkpoint_parallelsql.sbatch | 4 +-- comments_2_parquet_part2.py | 2 +- idf_authors.py | 43 ++++++++++++++++++++++++++ idf_comments.py | 58 +++++++++++++++++++++++++++++++++++ run_tf_jobs.sh | 2 +- tf_comments.py | 5 ++- 6 files changed, 106 insertions(+), 8 deletions(-) create mode 100644 idf_authors.py create mode 100644 idf_comments.py diff --git a/checkpoint_parallelsql.sbatch b/checkpoint_parallelsql.sbatch index a54aab1..1975802 100644 --- a/checkpoint_parallelsql.sbatch +++ b/checkpoint_parallelsql.sbatch @@ -10,11 +10,9 @@ ## Walltime (12 hours) #SBATCH --time=12:00:00 ## Memory per node -#SBATCH --mem=100G +#SBATCH --mem=32G #SBATCH --cpus-per-task=4 #SBATCH --ntasks=1 - - module load parallel_sql #Put here commands to load other modules (e.g. matlab etc.) diff --git a/comments_2_parquet_part2.py b/comments_2_parquet_part2.py index 7b17251..62580ac 100755 --- a/comments_2_parquet_part2.py +++ b/comments_2_parquet_part2.py @@ -26,4 +26,4 @@ df2.write.parquet("/gscratch/comdata/output/reddit_comments_by_subreddit.parquet df = df.repartition('author') df3 = df.sort(["author","CreatedAt","subreddit","link_id","parent_id","Year","Month","Day"],ascending=True) df3 = df3.sortWithinPartitions(["author","CreatedAt","subreddit","link_id","parent_id","Year","Month","Day"],ascending=True) -df3.write.parquet("/gscratch/comdata/output/reddit_comments_by_author.parquet", mode='overwrite') +df3.write.parquet("/gscratch/comdata/output/reddit_comments_by_author.parquet", mode='overwrite',compression='snappy') diff --git a/idf_authors.py b/idf_authors.py new file mode 100644 index 0000000..379de5a --- /dev/null +++ b/idf_authors.py @@ -0,0 +1,43 @@ +from pyspark.sql import functions as f +from pyspark.sql import SparkSession + +spark = SparkSession.builder.getOrCreate() +df = spark.read.parquet("/gscratch/comdata/users/nathante/reddit_tfidf_test_authors.parquet_temp/") + +max_subreddit_week_authors = df.groupby(['subreddit','week']).max('tf') +max_subreddit_week_authors = max_subreddit_week_authors.withColumnRenamed('max(tf)','sr_week_max_tf') + +df = df.join(max_subreddit_week_authors, ['subreddit','week']) + +df = df.withColumn("relative_tf", df.tf / df.sr_week_max_tf) + +# group by term / week +idf = df.groupby(['author','week']).count() + +idf = idf.withColumnRenamed('count','idf') + +# output: term | week | df +#idf.write.parquet("/gscratch/comdata/users/nathante/reddit_tfidf_test_sorted_tf.parquet_temp",mode='overwrite',compression='snappy') + +# collect the dictionary to make a pydict of terms to indexes +authors = idf.select('author').distinct() +authors = authors.withColumn('author_id',f.monotonically_increasing_id()) + + +# map terms to indexes in the tfs and the idfs +df = df.join(terms,on='author') + +idf = idf.join(terms,on='author') + +# join on subreddit/term/week to create tf/dfs indexed by term +df = df.join(idf, on=['author_id','week','author']) + +# agg terms by subreddit to make sparse tf/df vectors +df = df.withColumn("tf_idf",df.relative_tf / df.sr_week_max_tf) + +df = df.groupby(['subreddit','week']).agg(f.collect_list(f.struct('term_id','tf_idf')).alias('tfidf_maps')) + +df = df.withColumn('tfidf_vec', f.map_from_entries('tfidf_maps')) + +# output: subreddit | week | tf/df +df.write.parquet('/gscratch/comdata/users/nathante/test_tfidf_authors.parquet',mode='overwrite',compression='snappy') diff --git a/idf_comments.py b/idf_comments.py new file mode 100644 index 0000000..d29be80 --- /dev/null +++ b/idf_comments.py @@ -0,0 +1,58 @@ +from pyspark.sql import functions as f +from pyspark.sql import SparkSession + +spark = SparkSession.builder.getOrCreate() +df = spark.read.parquet("/gscratch/comdata/users/nathante/reddit_tfidf_test.parquet_temp") + +max_subreddit_week_terms = df.groupby(['subreddit','week']).max('tf') +max_subreddit_week_terms = max_subreddit_week_terms.withColumnRenamed('max(tf)','sr_week_max_tf') + +df = df.join(max_subreddit_week_terms, ['subreddit','week']) + +df = df.withColumn("relative_tf", df.tf / df.sr_week_max_tf) + +# group by term / week +idf = df.groupby(['term','week']).count() + +idf = idf.withColumnRenamed('count','idf') + +# output: term | week | df +#idf.write.parquet("/gscratch/comdata/users/nathante/reddit_tfidf_test_sorted_tf.parquet_temp",mode='overwrite',compression='snappy') + +# collect the dictionary to make a pydict of terms to indexes +terms = idf.select('term').distinct() +terms = terms.withColumn('term_id',f.monotonically_increasing_id()) + + +# print('collected terms') + +# terms = [t.term for t in terms] +# NTerms = len(terms) +# term_id_map = {term:i for i,term in enumerate(sorted(terms))} + +# term_id_map = spark.sparkContext.broadcast(term_id_map) + +# print('term_id_map is broadcasted') + +# def map_term(x): +# return term_id_map.value[x] + +# map_term_udf = f.udf(map_term) + +# map terms to indexes in the tfs and the idfs +df = df.join(terms,on='term') + +idf = idf.join(terms,on='term') + +# join on subreddit/term/week to create tf/dfs indexed by term +df = df.join(idf, on=['term_id','week','term']) + +# agg terms by subreddit to make sparse tf/df vectors +df = df.withColumn("tf_idf",df.relative_tf / df.sr_week_max_tf) + +df = df.groupby(['subreddit','week']).agg(f.collect_list(f.struct('term_id','tf_idf')).alias('tfidf_maps')) + +df = df.withColumn('tfidf_vec', f.map_from_entries('tfidf_maps')) + +# output: subreddit | week | tf/df +df.write.parquet('/gscratch/comdata/users/nathante/test_tfidf.parquet',mode='overwrite',compression='snappy') diff --git a/run_tf_jobs.sh b/run_tf_jobs.sh index fc191d4..0e7d5dd 100755 --- a/run_tf_jobs.sh +++ b/run_tf_jobs.sh @@ -1,6 +1,6 @@ #!/usr/bin/env bash module load parallel_sql -source ../bin/activate +source ./bin/activate python3 tf_comments.py gen_task_list psu --del --Y cat tf_task_list | psu --load diff --git a/tf_comments.py b/tf_comments.py index 211647e..cb3b628 100755 --- a/tf_comments.py +++ b/tf_comments.py @@ -161,9 +161,8 @@ def weekly_tf(partition, mwe_pass = 'first'): while True: chunk = islice(outrows,outchunksize) - chunk = (c for c in chunk if c.subreddit is not None) + chunk = (c for c in chunk if c[1] is not None) pddf = pd.DataFrame(chunk, columns=["is_token"] + schema.names) - author_pddf = pddf.loc[pddf.is_token == False, schema.names] pddf = pddf.loc[pddf.is_token == True, schema.names] author_pddf = author_pddf.rename({'term':'author'}, axis='columns') @@ -185,7 +184,7 @@ def gen_task_list(mwe_pass='first'): with open("tf_task_list",'w') as outfile: for f in files: if f.endswith(".parquet"): - outfile.write(f"python3 tf_comments.py weekly_tf --mwe-pass {mwe_pass} {f}\n") + outfile.write(f"./tf_comments.py weekly_tf --mwe-pass {mwe_pass} {f}\n") if __name__ == "__main__": fire.Fire({"gen_task_list":gen_task_list, -- 2.39.5