Compute IDF for terms and authors.

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.)

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')

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')

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')

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

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,