%md
# [SDS-2.x, Scalable Data Engineering Science](https://lamastex.github.io/scalable-data-science/sds/2/x/)

%md
# Word Count on US State of the Union (SoU) Addresses

* Word Count in big data is the equivalent of `Hello World` in programming
* We count the number of occurences of each word in the first and last (2016) SoU addresses.

**prerequisite** see **DO NOW** below. You should have loaded data as instructed in `scalable-data-science/xtraResources/sdsDatasets`.

#### DO NOW (if not done already)

In your databricks community edition:

1. In your `WorkSpace` create a Folder named `scalable-data-science`
2. `Import` the databricks archive file at the following URL:
    * [https://github.com/lamastex/scalable-data-science/raw/master/dbcArchives/2017/parts/xtraResources.dbc](https://github.com/lamastex/scalable-data-science/raw/master/dbcArchives/2017/parts/xtraResources.dbc)
3. This should open a structure of directories in with path: `/Workspace/scalable-data-science/xtraResources/`

%md

An interesting analysis of the textual content of the *State of the Union (SoU)* addresses by all US presidents was done in:

* [Alix Rule, Jean-Philippe Cointet, and Peter S. Bearman, Lexical shifts, substantive changes, and continuity in State of the Union discourse, 1790–2014, PNAS 2015 112 (35) 10837-10844; doi:10.1073/pnas.1512221112](http://www.pnas.org/content/112/35/10837.full).



![](http://www.pnas.org/content/112/35/10837/F5.large.jpg)

[Fig. 5](http://www.pnas.org/content/112/35/10837.full). A river network captures the flow across history of US political discourse, as perceived by contemporaries. Time moves along the x axis. Clusters on semantic networks of 300 most frequent terms for each of 10 historical periods are displayed as vertical bars. Relations between clusters of adjacent periods are indexed by gray flows, whose density reflects their degree of connection. Streams that connect at any point in history may be considered to be part of the same system, indicated with a single color. 

## Let us investigate this dataset ourselves!

1. We first get the source text data by scraping and parsing from [http://stateoftheunion.onetwothree.net/texts/index.html](http://stateoftheunion.onetwothree.net/texts/index.html) as explained in 
[scraping and parsing SoU addresses](/#workspace/scalable-data-science/xtraResources/sdsDatasets/scraperUSStateofUnionAddresses).
* This data is already made available in DBFS, our distributed file system.
* We only do the simplest word count with this data in this notebook and will do more sophisticated analyses in the sequel (including topic modeling, etc).

%md
## Project Suggestion

**Streaming/NLP/Vertex-Programs, etc**:

* [project: MEP - meme Evolution Programme](https://lamastex.github.io/scalable-data-science/sds/research/mep/) -  just won (2017-2018) AWS Cloud Computing Credits for research grant.
* [The GDELT Project: Watching our World Unfold](https://www.gdeltproject.org/)

%md
## Key Data Management Concepts 

### The Structure Spectrum

**(watch now 1:10)**:

[![Structure Spectrum by Anthony Joseph in BerkeleyX/CS100.1x](http://img.youtube.com/vi/pMSGGZVSwqo/0.jpg)](https://www.youtube.com/watch?v=pMSGGZVSwqo?rel=0&autoplay=1&modestbranding=1&start=1&end=70)

Here we will be working with **unstructured** or **schema-never** data (plain text files).
***

### Files

**(watch later 1:43)**:

[![Files by Anthony Joseph in BerkeleyX/CS100.1x](http://img.youtube.com/vi/NJyBQ-cQ3Ac/0.jpg)](https://www.youtube.com/watch?v=NJyBQ-cQ3Ac?rel=0&autoplay=1&modestbranding=1&start=1)

%md
###DBFS and dbutils - where is this dataset in our distributed file system?

* Since we are on the databricks cloud, it has a file system called DBFS
* DBFS is similar to HDFS, the Hadoop distributed file system
* dbutils allows us to interact with dbfs.
* The 'display' command displays the list of files in a given directory in the file system.

%md
Let us display the *head* or the first few lines of the file `dbfs:/datasets/sou/17900108.txt` to see what it contains using `dbutils.fs.head` method.  
`head(file: String, maxBytes: int = 65536): String` -> Returns up to the first 'maxBytes' bytes of the given file as a String encoded in UTF-8
as follows:

%md
##### You Try!
Uncomment and modify `xxxx` in the cell below to read the first 1000 bytes from the file.

%md
### Read the file into Spark Context as an RDD of Strings

* The `textFile` method on the available `SparkContext` `sc` can read the text file `dbfs:/datasets/sou/17900108.txt` into Spark and create an RDD of Strings
  * but this is done lazily until an action is taken on the RDD `sou17900108`!

%md
### Perform some actions on the RDD

* Each String in the RDD `sou17900108` represents one line of data from the file and can be made to perform one of the following actions:
  * count the number of elements in the RDD `sou17900108` (i.e., the number of lines in the text file `dbfs:/datasets/sou/17900108.txt`) using `sou17900108.count()`
  * display the contents of the RDD using `take` or `collect`.

%md
### Cache the RDD in (distributed) memory to avoid recreating it for each action

* Above, every time we took an action on the same RDD, the RDD was reconstructed from the textfile.  
  * Spark's advantage compared to Hadoop MapReduce is the ability to cache or store the RDD in distributed memory across the nodes.
* Let's use `.cache()` after creating an RDD so that it is in memory after the first action (and thus avoid reconstruction for subsequent actions).
  * count the number of elements in the RDD `sou17900108` (i.e., the number of lines in the text file `dbfs:/datasets/sou/17900108.txt`) using `sou17900108.count()`
  * display the contents of the RDD using `take` or `collect`.

%md
#### Lifecycle of a Spark Program
    
**(watch now 0:23)**:

[![Spark Program Lifecycle by Anthony Joseph in BerkeleyX/CS100.1x](http://img.youtube.com/vi/HWZUqNYAJj4/0.jpg)](https://www.youtube.com/watch?v=HWZUqNYAJj4?rel=0&autoplay=1&modestbranding=1&start=1)

##### Summary

* create RDDs from:
  * some external data source (such as a distributed file system)
  * parallelized collection in your driver program
* lazily transform these RDDs into new RDDs
* cache some of those RDDs for future reuse
* you perform actions to execute parallel computation to produce results

%md
### Transform lines to words

* We need to loop through each line and split the line into words
* For now, let us split using whitespace
* More sophisticated regular expressions can be used to split the line (as we will see soon)

%md
### Naive word count

At a first glace, to do a word count of George Washingtons SoU address, we are templed to do the following:

 * just break each line by the whitespace character " " and find the words using a `flatMap`
 * then do the `map` with the closure `word => (word, 1)` to initialize each `word` with a integer count of `1` 
    * ie., transform each word to a *(key, value)* pair or `Tuple` such as `(word, 1)`
 * then count all *value*s with the same *key* (`word` is the Key in our case) by doing a
   * `reduceByKey(_+_)` 
 * and finally `collect()` to display the results.

%md
Unfortunately, as you can see from the `collect` above:

* the words have punctuations at the end which means that the same words are being counted as different words. Eg: importance
* empty words are being counted

So we need a bit of `regex`'ing or regular-expression matching (all readily available from Scala via Java String types).

We will cover the three things we want to do with a simple example from Middle Earth!

* replace all multiple whitespace characters with one white space character " "
* replace all punction characters we specify within `[` and `]` such as `[,?.!:;]` by the empty string `""` (i.e., remove these punctuation characters)
* convert everything to lower-case.

%md
### More sophisticated word count
We are now ready to do a word count of George Washington's SoU on January 8th 1790 as follows: