Refactor and reorganze.

ngrams/checkpoint_parallelsql.sbatch

@@ -0,0 +1,26 @@
+#!/bin/bash
+## parallel_sql_job.sh
+#SBATCH --job-name=tf_subreddit_comments
+## Allocation Definition
+#SBATCH --account=comdata-ckpt
+#SBATCH --partition=ckpt
+## Resources
+## Nodes. This should always be 1 for parallel-sql.
+#SBATCH --nodes=1    
+## Walltime (12 hours)
+#SBATCH --time=12:00:00
+## Memory per node
+#SBATCH --mem=32G
+#SBATCH --cpus-per-task=4
+#SBATCH --ntasks=1
+#SBATCH -D /gscratch/comdata/users/nathante/cdsc-reddit
+source ./bin/activate
+module load parallel_sql
+echo $(which perl)
+conda list pyarrow
+which python3
+#Put here commands to load other modules (e.g. matlab etc.)
+#Below command means that parallel_sql will get tasks from the database
+#and run them on the node (in parallel). So a 16 core node will have
+#16 tasks running at one time.
+parallel-sql --sql -a parallel --exit-on-term --jobs 4

ngrams/run_tf_jobs.sh

@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+module load parallel_sql
+source ./bin/activate
+python3 tf_comments.py gen_task_list
+psu --del --Y
+cat tf_task_list | psu --load
+
+for job in $(seq 1 50); do sbatch checkpoint_parallelsql.sbatch; done;

ngrams/sort_tf_comments.py

@@ -0,0 +1,13 @@
+#!/usr/bin/env python3
+
+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/")
+
+df = df.repartition(2000,'term')
+df = df.sort(['term','week','subreddit'])
+df = df.sortWithinPartitions(['term','week','subreddit'])
+
+df.write.parquet("/gscratch/comdata/users/nathante/reddit_tfidf_test_sorted_tf.parquet_temp",mode='overwrite',compression='snappy')

ngrams/tf_comments.py

@@ -0,0 +1,195 @@
+#!/usr/bin/env python3
+import pandas as pd
+import pyarrow as pa
+import pyarrow.dataset as ds
+import pyarrow.parquet as pq
+from itertools import groupby, islice, chain
+import fire
+from collections import Counter
+import os
+import re
+from nltk import wordpunct_tokenize, MWETokenizer, sent_tokenize
+from nltk.corpus import stopwords
+from nltk.util import ngrams
+import string
+from random import random
+
+# remove urls
+# taken from https://stackoverflow.com/questions/3809401/what-is-a-good-regular-expression-to-match-a-url
+urlregex = re.compile(r"[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b([-a-zA-Z0-9()@:%_\+.~#?&//=]*)")
+
+# compute term frequencies for comments in each subreddit by week
+def weekly_tf(partition, mwe_pass = 'first'):
+    dataset = ds.dataset(f'/gscratch/comdata/output/reddit_comments_by_subreddit.parquet/{partition}', format='parquet')
+    if not os.path.exists("/gscratch/comdata/users/nathante/reddit_comment_ngrams_10p_sample/"):
+        os.mkdir("/gscratch/comdata/users/nathante/reddit_comment_ngrams_10p_sample/")
+
+    if not os.path.exists("/gscratch/comdata/users/nathante/reddit_tfidf_test_authors.parquet_temp/"):
+        os.mkdir("/gscratch/comdata/users/nathante/reddit_tfidf_test_authors.parquet_temp/")
+
+    ngram_output = partition.replace("parquet","txt")
+
+    if mwe_pass == 'first':
+        if os.path.exists(f"/gscratch/comdata/output/reddit_ngrams/comment_ngrams_10p_sample/{ngram_output}"):
+            os.remove(f"/gscratch/comdata/output/reddit_ngrams/comment_ngrams_10p_sample/{ngram_output}")
+    
+    batches = dataset.to_batches(columns=['CreatedAt','subreddit','body','author'])
+
+
+    schema = pa.schema([pa.field('subreddit', pa.string(), nullable=False),
+                        pa.field('term', pa.string(), nullable=False),
+                        pa.field('week', pa.date32(), nullable=False),
+                        pa.field('tf', pa.int64(), nullable=False)]
+    )
+
+    author_schema = pa.schema([pa.field('subreddit', pa.string(), nullable=False),
+                               pa.field('author', pa.string(), nullable=False),
+                               pa.field('week', pa.date32(), nullable=False),
+                               pa.field('tf', pa.int64(), nullable=False)]
+    )
+
+    dfs = (b.to_pandas() for b in batches)
+
+    def add_week(df):
+        df['week'] = (df.CreatedAt - pd.to_timedelta(df.CreatedAt.dt.dayofweek, unit='d')).dt.date
+        return(df)
+
+    dfs = (add_week(df) for df in dfs)
+
+    def iterate_rows(dfs):
+        for df in dfs:
+            for row in df.itertuples():
+                yield row
+
+    rows = iterate_rows(dfs)
+
+    subreddit_weeks = groupby(rows, lambda r: (r.subreddit, r.week))
+
+    if mwe_pass != 'first':
+        mwe_dataset = pd.read_feather(f'/gscratch/comdata/output/reddit_ngrams/multiword_expressions.feather')
+        mwe_dataset = mwe_dataset.sort_values(['phrasePWMI'],ascending=False)
+        mwe_phrases = list(mwe_dataset.phrase)
+        mwe_phrases = [tuple(s.split(' ')) for s in mwe_phrases]
+        mwe_tokenizer = MWETokenizer(mwe_phrases)
+        mwe_tokenize = mwe_tokenizer.tokenize
+    
+    else:
+        mwe_tokenize = MWETokenizer().tokenize
+
+    def remove_punct(sentence):
+        new_sentence = []
+        for token in sentence:
+            new_token = ''
+            for c in token:
+                if c not in string.punctuation:
+                    new_token += c
+            if len(new_token) > 0:
+                new_sentence.append(new_token)
+        return new_sentence
+
+    stopWords = set(stopwords.words('english'))
+
+    # we follow the approach described in datta, phelan, adar 2017
+    def my_tokenizer(text):
+        # remove stopwords, punctuation, urls, lower case
+        # lowercase        
+        text = text.lower()
+
+        # remove urls
+        text = urlregex.sub("", text)
+
+        # sentence tokenize
+        sentences = sent_tokenize(text)
+
+        # wordpunct_tokenize
+        sentences = map(wordpunct_tokenize, sentences)
+
+        # remove punctuation
+                        
+        sentences = map(remove_punct, sentences)
+
+        # remove sentences with less than 2 words
+        sentences = filter(lambda sentence: len(sentence) > 2, sentences)
+
+        # datta et al. select relatively common phrases from the reddit corpus, but they don't really explain how. We'll try that in a second phase.
+        # they say that the extract 1-4 grams from 10% of the sentences and then find phrases that appear often relative to the original terms
+        # here we take a 10 percent sample of sentences 
+        if mwe_pass == 'first':
+            sentences = list(sentences)
+            for sentence in sentences:
+                if random() <= 0.1:
+                    grams = list(chain(*map(lambda i : ngrams(sentence,i),range(4))))
+                    with open(f'/gscratch/comdata/output/reddit_ngrams/comment_ngrams_10p_sample/{ngram_output}','a') as gram_file:
+                        for ng in grams:
+                            gram_file.write(' '.join(ng) + '\n')
+                for token in sentence:
+                    if token not in stopWords:
+                        yield token
+
+        else:
+            # remove stopWords
+            sentences = map(mwe_tokenize, sentences)
+            sentences = map(lambda s: filter(lambda token: token not in stopWords, s), sentences)
+            for sentence in sentences:
+                for token in sentence:
+                    yield token
+
+    def tf_comments(subreddit_weeks):
+        for key, posts in subreddit_weeks:
+            subreddit, week = key
+            tfs = Counter([])
+            authors = Counter([])
+            for post in posts:
+                tokens = my_tokenizer(post.body)
+                tfs.update(tokens)
+                authors.update([post.author])
+
+            for term, tf in tfs.items():
+                yield [True, subreddit, term, week, tf]
+
+            for author, tf in authors.items():
+                yield [False, subreddit, author, week, tf]
+
+    outrows = tf_comments(subreddit_weeks)
+
+    outchunksize = 10000
+
+    with pq.ParquetWriter(f"/gscratch/comdata/output/reddit_ngrams/comment_terms.parquet/{partition}",schema=schema,compression='snappy',flavor='spark') as writer, pq.ParquetWriter(f"/gscratch/comdata/output/reddit_ngrams/comment_authors.parquet/{partition}",schema=author_schema,compression='snappy',flavor='spark') as author_writer:
+    
+        while True:
+
+            chunk = islice(outrows,outchunksize)
+            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')
+            author_pddf = author_pddf.loc[:,author_schema.names]
+            table = pa.Table.from_pandas(pddf,schema=schema)
+            author_table = pa.Table.from_pandas(author_pddf,schema=author_schema)
+            do_break = True
+
+            if table.shape[0] != 0:
+                writer.write_table(table)
+                do_break = False
+            if author_table.shape[0] != 0:
+                author_writer.write_table(author_table)
+                do_break = False
+
+            if do_break:
+                break
+
+        writer.close()
+        author_writer.close()
+
+
+def gen_task_list(mwe_pass='first'):
+    files = os.listdir("/gscratch/comdata/output/reddit_comments_by_subreddit.parquet/")
+    with open("tf_task_list",'w') as outfile:
+        for f in files:
+            if f.endswith(".parquet"):
+                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,
+               "weekly_tf":weekly_tf})

ngrams/top_comment_phrases.py

@@ -0,0 +1,58 @@
+from pyspark.sql import functions as f
+from pyspark.sql import Window
+from pyspark.sql import SparkSession
+import numpy as np
+
+spark = SparkSession.builder.getOrCreate()
+df = spark.read.text("/gscratch/comdata/users/nathante/reddit_comment_ngrams_10p_sample/")
+
+df = df.withColumnRenamed("value","phrase")
+
+# count phrase occurrances
+phrases = df.groupby('phrase').count()
+phrases = phrases.withColumnRenamed('count','phraseCount')
+phrases = phrases.filter(phrases.phraseCount > 10)
+
+
+# count overall
+N = phrases.select(f.sum(phrases.phraseCount).alias("phraseCount")).collect()[0].phraseCount
+
+print(f'analyzing PMI on a sample of {N} phrases') 
+logN = np.log(N)
+phrases = phrases.withColumn("phraseLogProb", f.log(f.col("phraseCount")) - logN)
+
+# count term occurrances
+phrases = phrases.withColumn('terms',f.split(f.col('phrase'),' '))
+terms = phrases.select(['phrase','phraseCount','phraseLogProb',f.explode(phrases.terms).alias('term')])
+
+win = Window.partitionBy('term')
+terms = terms.withColumn('termCount',f.sum('phraseCount').over(win))
+terms = terms.withColumnRenamed('count','termCount')
+terms = terms.withColumn('termLogProb',f.log(f.col('termCount')) - logN)
+
+terms = terms.groupBy(terms.phrase, terms.phraseLogProb, terms.phraseCount).sum('termLogProb')
+terms = terms.withColumnRenamed('sum(termLogProb)','termsLogProb')
+terms = terms.withColumn("phrasePWMI", f.col('phraseLogProb') - f.col('termsLogProb'))
+
+# join phrases to term counts
+
+
+df = terms.select(['phrase','phraseCount','phraseLogProb','phrasePWMI'])
+
+df = df.sort(['phrasePWMI'],descending=True)
+df = df.sortWithinPartitions(['phrasePWMI'],descending=True)
+df.write.parquet("/gscratch/comdata/users/nathante/reddit_comment_ngrams_pwmi.parquet/",mode='overwrite',compression='snappy')
+
+df = spark.read.parquet("/gscratch/comdata/users/nathante/reddit_comment_ngrams_pwmi.parquet/")
+
+df.write.csv("/gscratch/comdata/users/nathante/reddit_comment_ngrams_pwmi.csv/",mode='overwrite',compression='none')
+
+df = spark.read.parquet("/gscratch/comdata/users/nathante/reddit_comment_ngrams_pwmi.parquet")
+df = df.select('phrase','phraseCount','phraseLogProb','phrasePWMI')
+
+# choosing phrases occurring at least 3500 times in the 10% sample (35000 times) and then with a PWMI of at least 3 yeids about 65000 expressions.
+#
+df = df.filter(f.col('phraseCount') > 3500).filter(f.col("phrasePWMI")>3)
+df = df.toPandas()
+df.to_feather("/gscratch/comdata/users/nathante/reddit_multiword_expressions.feather")
+df.to_csv("/gscratch/comdata/users/nathante/reddit_multiword_expressions.csv")