<jupyter_start><jupyter_text>Spark Dataframes (and RDDs) for manipulating text In your cirrus account, in the login node, type the following commands, for installing different nltk packages that we will use later.  ```[XXX@cirrus-login0 lab_exercises]$ pip install nltk[XXX@cirrus-login0 lab_exercises]$ pythonPython 3.6.4 |Anaconda, Inc.| (default, Jan 16 2018, 18:10:19) [GCC 7.2.0] on linuxType "help", "copyright", "credits" or "license" for more information.>>> import nltk>>> nltk.download('wordnet')>>> nltk.download('punkt')>>> nltk.download('stopwords')``` The aim of this exercise is to learn how to manipulate data using spark dataframes. We are going to work with a subset of the encyclopaedia britannica from the [National Library of Scotland](https://data.nls.uk/data/digitised-collections/encyclopaedia-britannica/). We have downloaded previously the full dataset, ingested the data, and produce a subsample dataset, called *"nls_demo.csv"* (which is the one we are going to use) using a spark-tool called [defoe](https://github.com/alan-turing-institute/defoe/blob/master/docs/nls_demo_examples/nls_demo_individual_queries.md).<jupyter_code>import nltk
import string<jupyter_output><empty_output><jupyter_text>This is the format of the CSV file:title,edition,year,place,archive_filename,page_filename,page_id,num_pages,type_archive,model,preprocess,page_string<jupyter_code>#Reading the csv file into a dataframe
df= sqlContext.read.csv("/lustre/home/shared/y15/spark/data/nls_demo.csv", header="true")
df.show(3)
#Filtering by pages that are not null, and grouping by year, and counting the number of pages
df.filter(df.page_string.isNotNull()).select(df.year, df.page_string).groupby(df.year).count().show()
# Checking how many rows do we have with the value "year" in the column "year"
df[df.year.like("year")].collect()
# Filter again the data, which pages are not null, and which years are not "year", and selecting 2 columns, and counting the elements by year
df.filter(df.page_string.isNotNull()).filter(df["year"]!="year").select(df.year, df.page_string).groupby(df.year).count().show()
#Same that before, but grouping by place
df.filter(df.page_string.isNotNull()).filter(df["year"]!="year").select(df.place, df.page_string).groupby(df.place).count().show()
#Grouping by years, but just the ones between 1773 and 1842
df.filter(df.page_string.isNotNull()).filter(df["year"]!="year").filter(df.year.between(1773, 1842)).select(df.place, df.edition, df.page_string, df.year).groupby(df.year).count().show()
df.filter(df.page_string.isNotNull()).filter(df["year"]!="year").filter(df.year.between(1773, 1842)).select(df.place, df.edition, df.page_string, df.year).show()
df.filter(df.page_string.isNotNull()).filter(df["year"]!="year").filter(df.year.between(1773, 1842)).filter(df.edition.startswith("Second")).select(df.place, df.edition, df.page_string, df.year).show()
# Now lets create a datafame, which pages are not Null, and selecting just year and page_string columns
newdf=df.filter(df.page_string.isNotNull()).select(df.year, df.page_string)
# And check the Schema of the new dataframe
newdf.printSchema()
# Count the number of rows
newdf.count()
# Show the first 20 rows
newdf.show()
# Converting the dataframe to tuples- best suited for processing unstructured data.
pages=newdf.rdd.map(tuple)
pages.take(8)
def sent_TokenizeFunct(x):
    print ("%s" %x)
    return nltk.sent_tokenize(x)
sentenceTokenizeRDD = pages.map(lambda p: sent_TokenizeFunct(p[1]))
sentenceTokenizeRDD.take(5)
def word_TokenizeFunct(x):
    splitted = [word for line in x for word in line.split()]
    return splitted
wordTokenizeRDD = sentenceTokenizeRDD.map(word_TokenizeFunct)
wordTokenizeRDD.take(5)
def removeStopWordsFunct(x):
    nltk.download('stopwords')
    from nltk.corpus import stopwords
    stop_words=set(stopwords.words('english'))
    filteredSentence = [w for w in x if not w in stop_words]
    return filteredSentence<jupyter_output><empty_output><jupyter_text>stopwordRDD = wordTokenizeRDD.map(removeStopWordsFunct)<jupyter_code>stopwordRDD.take(10)
def removePunctuationsFunct(x):
    nltk.download('punkt')
    list_punct=list(string.punctuation)
    filtered = [''.join(c for c in s if c not in list_punct) for s in x] 
    filtered_space = [s for s in filtered if s] #remove empty space 
    return filtered
rmvPunctRDD = stopwordRDD.map(removePunctuationsFunct)
rmvPunctRDD.take(10)
def lemmatizationFunct(x):
    nltk.download('wordnet')
    lemmatizer = nltk.WordNetLemmatizer()
    finalLem = [lemmatizer.lemmatize(s) for s in x]
    return finalLem
lem_wordsRDD = rmvPunctRDD.map(lemmatizationFunct)
lem_wordsRDD.take(10)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12817231819151204, 'fraction_numerical': 0.014502026018340797, 'mean_word_length': 6.149390243902439, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 25, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '334767', 'license': 'Apache-2.0', 'n_tokens_mistral': 1653, 'n_tokens_neox': 1575, 'n_words': 397, 'path': 'lab_exercises/lab_rdd_dataframes_NLTK.ipynb', 'repo_name': 'masonlr/prace-spark-for-data-scientists', 'size': '4689'}