# Adrian deWynter, 2016
'''
Adrian deWynter (2016)
Notebook corresponding to an Apache Spark class I once took.
This one pertains to analysis of texts, more specifically word count.
'''

#####
# Remember, databricks has a built-in function (display) that isn't available elsewhere.
# This code isn't meant to run anywhere that isn't Spark -- and some databricks functions
# may still be lunging around.
# I removed testing code and most of the stuff that could be used to correctly identify
# this file when someone is looking up the answers.
#
# Index:
#   ------
#   1 - Basic string operations
#   2 - DF operations with words
#   3 - Shakesepearian Analysis
#####


'''
#############################
# 1 - Basic string operations
#############################
'''
# Perform an operation that adds an 's' to each word in a df
wordsDF = sqlContext.createdf([('cat',), ('elephant',), ('rat',), ('rat',), ('cat', )], ['word'])
wordsDF.show()
print type(wordsDF)
wordsDF.printSchema()

from pyspark.sql.functions import lit, concat
pluralDF = wordsDF.select(concat(wordsDF.word, lit('s')).alias('word'))
pluralDF.show()
 
# Use the SQL length function to find the number of characters in each word
from pyspark.sql.functions import length
pluralLengthsDF = pluralDF.select(length('word'))
pluralLengthsDF.show()

'''
##############################
# 2 - DF operations with words
##############################
'''
# Find the counts of words and the number of times that these words occur.
wordCountsDF = (wordsDF.groupBy(wordsDF.word).count())
wordCountsDF.show()

# Calculate the number of unique words in wordsDF
from spark_notebook_helpers import printdfs

#This function returns all the dfs in the notebook and their corresponding column names.
printdfs(True)
uniqueWordsCount = wordCountsDF.groupBy(wordCountsDF.word).count()
uniqueWordsCount = uniqueWordsCount.count()
print uniqueWordsCount

# Find the mean number of occurrences of words in wordCountsDF.
averageCount = (wordCountsDF.groupBy().mean('count')).head()[0]
print averageCount

# Creates a df with word counts.
# Args:  wordListDF (df of str): A df consisting of one string column called 'word'.
# Returns df of (str, int): A df containing 'word' and 'count' columns.
def wordCount(wordListDF):
    return wordListDF.groupBy(wordListDF.word).count()

wordCount(wordsDF).show()


from pyspark.sql.functions import regexp_replace, trim, col, lower

# Removes punctuation, changes to lower case, and strips leading and trailing spaces.
# Args: a Column containing a sentence.
# Returns a Column named 'sentence' with clean-up operations applied.
def removePunctuation(column):
    
    return lower(trim(regexp_replace(column, "[^0-9a-zA-Z ]", ""))).alias("sentence")

sentenceDF = sqlContext.createdf([('Hi, you!',),
                                         (' No under_score!',),
                                         (' *      Remove punctuation then spaces  * ',)], ['sentence'])
sentenceDF.show(truncate=False)
(sentenceDF
 .select(removePunctuation(col('sentence')))
 .show(truncate=False))


'''
#############################
# 1 - Shakesepearian Analysis
#############################
'''
# Use http://www.gutenberg.org/ebooks/100
fileName = ""
shakespeareDF = sqlContext.read.text(fileName).select(removePunctuation(col('value')))
shakespeareDF.show(15, truncate=False)

# Split each 'sentence' in the df by its spaces
# Transform from a df that contains lists of words into a df with each word in its own row.
# Remove the rows that contain ''.
from pyspark.sql.functions import split, explode
shakeWordsDF = shakespeareDF.select(split(shakespeareDF.sentence, ' ').alias("sentence"))
shakeWordsDF = shakeWordsDF.select(explode(shakeWordsDF.sentence).alias("word"))
shakeWordsDF = shakeWordsDF.filter("word != ''")

shakeWordsDF.show()
shakeWordsDFCount = shakeWordsDF.count()
print shakeWordsDFCount

# Apply the wordCount() function to produce a list of word counts.
from pyspark.sql.functions import desc
topWordsAndCountsDF = wordCount(shakeWordsDF).orderBy("count", ascending=False)
topWordsAndCountsDF.show()