Compute IDF for terms and authors.

diff --git a/checkpoint_parallelsql.sbatch b/checkpoint_parallelsql.sbatch
index a54aab124d2b83b2dd65a8bbdc4cd9d255e1ee0f..1975802daa4b26745d5e917fe6acde46a9303de6 100644 (file)
--- 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 7b17251fdb9e34096e4e2ac46dcba470a9823357..62580acf605e92c39a4f93c5f15f272d612b8e41 100755 (executable)
--- 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 (file)
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 (file)
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 fc191d48de8edd95d72e79b06baf175c647dd6fe..0e7d5dd229f0a8c1a3d58255c6f9dd82e2856b18 100755 (executable)
--- 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 211647e11521052390b4c0c754797cc2c55e651a..cb3b6288c37079b20f5355cbdf54d28bf1f29495 100755 (executable)
--- 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,