<repo_name>davidgabriel42/CS691_PA2<file_sep>/q3/q3.py
import pandas as pd
import numpy as np
import re
import findspark
findspark.init('/home/dave/spark-2.4.1-bin-hadoop2.7/')
import pyspark
from pyspark.sql.functions import *
from pyspark.sql.types import *
#import regexp_extract, col, lit
from pyspark.sql import *
import sys
import os

fn1 = sys.argv[1]
if os.path.exists(fn1):
        tweetsFile = fn1
fn2 = sys.argv[2]
if os.path.exists(fn2):
        usersFile = fn2

spark = SparkSession.builder.appName("Q1").getOrCreate()
tweets = spark.read.text(tweetsFile)
users = spark.read.text(usersFile)

#split DF into Cols: uid, tid, tweet
split_col = pyspark.sql.functions.split(tweets['value'], '\t')
tweets = tweets.withColumn('uid', split_col.getItem(0))
tweets = tweets.withColumn('tid', split_col.getItem(1))
tweets = tweets.withColumn('tweet', split_col.getItem(2))
split_col = pyspark.sql.functions.split(tweets['value'], '(\t)(?:20)')
tweets = tweets.withColumn('date', split_col.getItem(1))

tweets = tweets.select('uid', 'tid', 'tweet', 'date')
#tweets.show()

#process users to DF
split_users = pyspark.sql.functions.split(users['value'], '\t')
users = users.withColumn('uid', split_users.getItem(0))
users = users.withColumn('loc', split_users.getItem(1))
users = users.select('uid','loc')

LAPattern = '(Los\sAngeles)'
#extract mentions and create new DF

users = users.filter(users.loc.rlike(LAPattern))
#users.printSchema()
#users.show(n=20)

#09/16/2009 - 09/20/2009 extract tweets from these dates
datePattern = '(09-09-)(16|17|18|19|20)'
tweets = tweets.filter(tweets.date.rlike(datePattern))
#tweets.show()

#join on user id
users = users.select(col('uid').alias('uid1'), col('loc'))
join = tweets.join(users, tweets.uid == users.uid1)
#join.show()

LA = join.groupBy("uid").count().sort(desc("count"))

#redirect stdio and print result to file
sys.stdout = open('most_tweeted_users.txt', 'w')
print("Q3: 10 most tweeted users by uid from LA 9/16-9/20/2009")
LA.show(n=10)

spark.stop()
<file_sep>/readme.txt
       (                     (   (       )                                
   (   )\ ) (       )   )    )\ ))\ ) ( /(             (    *   )      )  
   )\ (()/( )\ ) ( /(( /(   (()/(()/( )\())   (  (     )\ ` )  /(   ( /(  
 (((_) /(_)|()/( )\())\())__ /(_))(_)|(_)\    )\ )\  (((_) ( )(_))  )(_)) 
 )\___(_))  /(_)|(_)((_)\___(_))(_))   ((_)  ((_|(_) )\___(_(_())  ((_)   
((/ __/ __|(_) // _(_) (_)  | _ \ _ \ / _ \ _ | | __((/ __|_   _|  |_  )  
 | (__\__ \ / _ \_, /| |    |  _/   /| (_) | || | _| | (__  | |     / /   
  \___|___/ \___//_/ |_|    |_| |_|_\ \___/ \__/|___| \___| |_|    /___|  
                                                                          
CS691 PROJECT 2
==========================================================
Set up:
I followed this to set up Jupyter, which also set up PySpark (included).
I think Jupyter is an excellent tool for working out problems once you
get used to it. I highly recommend it.

https://blog.sicara.com/get-started-pyspark-jupyter-guide-tutorial-ae2fe84f594f

I did not have to do anything else to run the scripts through the Python
interpreter. 
==========================================================
How to run: 
All .py files can be run from the python interpreter as follows:

$python q1.py /[PATHTOFILE]/training_set_tweets.txt /[PATHTOFILE]/training_set_users.txt
==========================================================
Versions Used to Build:
#python
$python --version
Python 3.6.5 :: Anaconda, Inc.

#spark
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.1
      /_/
         
Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_201)
Type in expressions to have them evaluated.
Type :help for more information.


#java (does it matter? probably)
$ java -version
java version "1.8.0_201"
Java(TM) SE Runtime Environment (build 1.8.0_201-b09)
Java HotSpot(TM) 64-Bit Server VM (build 25.201-b09, mixed mode)

<file_sep>/q2/q2.py
import pandas as pd
import numpy as np
import re
import findspark
findspark.init('/home/dave/spark-2.4.1-bin-hadoop2.7/')
import pyspark
from pyspark.sql.functions import *
from pyspark.sql.types import *
#import regexp_extract, col, lit
from pyspark.sql import *
import sys
import os

fn1 = sys.argv[1]
if os.path.exists(fn1):
        tweetsFile = fn1
fn2 = sys.argv[2]
if os.path.exists(fn2):
        usersFile = fn2

spark = SparkSession.builder.appName("q2").getOrCreate()
tweets = spark.read.text(tweetsFile)
users = spark.read.text(usersFile)

#split DF into Cols: uid, tid, tweet
split_col = pyspark.sql.functions.split(tweets['value'], '\t')
tweets = tweets.withColumn('uid', split_col.getItem(0))
tweets = tweets.withColumn('tid', split_col.getItem(1))
tweets = tweets.withColumn('tweet', split_col.getItem(2))
split_col = pyspark.sql.functions.split(tweets['value'], '(\t)(?:20)')
tweets = tweets.withColumn('date', split_col.getItem(1))

tweets = tweets.select('uid', 'tid', 'tweet', 'date')
#tweets.show()

#regex pattern for retweets
RTPattern = '(?:RT\s)(@\w+)'

#extract retweets and create new DF
retweets = tweets.withColumn('Retweets', regexp_extract(col('tweet'), RTPattern, 1))
retweets = retweets.groupBy("Retweets").count().sort(desc("count"))

#clear null tuples
retweets = retweets.filter(retweets.Retweets.rlike('@'))

#redirect stdio and print result to file
sys.stdout = open('most_retweeted_users.txt', 'w')

print("Q2: 10 most retweeted users")
retweets.show(n=10)

spark.stop()
<file_sep>/q1/q1.py
import pandas as pd
import numpy as np
import re
import findspark
findspark.init('/home/dave/spark-2.4.1-bin-hadoop2.7/')
import pyspark
from pyspark.sql.functions import *
from pyspark.sql.types import *
#import regexp_extract, col, lit
from pyspark.sql import *
import sys
import os

fn1 = sys.argv[1]
if os.path.exists(fn1):
	tweetsFile = fn1
fn2 = sys.argv[2]
if os.path.exists(fn2):
	usersFile = fn2

spark = SparkSession.builder.appName("q1").getOrCreate()
tweets = spark.read.text(tweetsFile)
users = spark.read.text(usersFile)

#split DF into Cols: uid, tid, tweet
split_col = pyspark.sql.functions.split(tweets['value'], '\t')
tweets = tweets.withColumn('uid', split_col.getItem(0))
tweets = tweets.withColumn('tid', split_col.getItem(1))
tweets = tweets.withColumn('tweet', split_col.getItem(2))
split_col = pyspark.sql.functions.split(tweets['value'], '(\t)(?:20)')
tweets = tweets.withColumn('date', split_col.getItem(1))

tweets = tweets.select('uid', 'tid', 'tweet', 'date')
#tweets.show()

#regex pattern for mentions
mentionPattern = '(?<!RT\s)(@\w+)'

#extract mentions and create new DF
mentions = tweets.withColumn('Mentions', regexp_extract(col('tweet'), mentionPattern, 1))
mentions = mentions.groupBy("Mentions").count().sort(desc("count"))

#clear null tuples
mentions = mentions.filter(mentions.Mentions.rlike('@'))

#redirect stdout and print results
sys.stdout = open('popular_mentions.txt', 'w')
print("Q1: 20 most mentions")
mentions.show(n=20)

spark.stop()
