%md
# [SDS-2.2-360-in-525-01: Intro to Apache Spark for data Scientists](https://lamastex.github.io/scalable-data-science/360-in-525/2018/01/)
### [SDS-2.2, Scalable Data Science](https://lamastex.github.io/scalable-data-science/sds/2/2/)

%md
This is an elaboration of the [http://spark.apache.org/docs/latest/sql-programming-guide.html](http://spark.apache.org/docs/latest/sql-programming-guide.html) by Ivan Sadikov and Raazesh Sainudiin.

# Getting Started
## Spark Sql Programming Guide

-   Getting Started
    -   Starting Point: SQLContext
    -   Creating DataFrames
    -   DataFrame Operations
    -   Running SQL Queries Programmatically
    -   Creating Datasets
    -   Interoperating with RDDs
        -   Inferring the Schema Using Reflection
        -   Programmatically Specifying the Schema

%md

# Getting Started

## Starting Point: SQLContext

The entry point into all functionality in Spark SQL is the
[`SparkSession`](http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.SparkSession) class and/or `SQLContext`/`HiveContext`. Spark session is created for you as `spark` when you start **spark-shell** or **pyspark**. You will need to create `SparkSession` usually when building an application (running on production-like on-premises cluster). n this case follow code below to create Spark session.


>import org.apache.spark.sql.SparkSession
>
>val spark = SparkSession.builder().appName("Spark SQL basic example").getOrCreate()
>
>// you could get SparkContext and SQLContext from SparkSession
>val sc = spark.sparkContext
>val sqlContext = spark.sqlContext
>
>// This is used to implicitly convert an RDD or Seq to a DataFrame (see examples below)
>import spark.implicits._


But in Databricks notebook (similar to `spark-shell`) `SparkSession` is already created for you and is available as `spark`.

%md
After evaluation you should see something like this:
```
res0: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@2d0c6c9
```
In order to enable Hive support use `enableHiveSupport()` method on builder when constructing Spark session, which provides richer functionality over standard Spark SQL context, for example, usage of Hive user-defined functions or loading and writing data from/into Hive. Note that most of the SQL functionality is available regardless Hive support.

%md
## Creating DataFrames

With a `SparkSessions`, applications can create [Dataset](https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.Dataset) or [`DataFrame`](https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.DataFrame) from an [existing `RDD`](sql-programming-guide.html#interoperating-with-rdds), from a Hive table, or from various [datasources](sql-programming-guide.html#data-sources).

Just to recap, a DataFrame is a distributed collection of data organized into named columns. You can think of it as an organized into table RDD of case class `Row` (which is not exactly true). DataFrames, in comparison to RDDs, are backed by rich optimizations, including tracking their own schema, adaptive query execution, code generation including whole stage codegen, extensible Catalyst optimizer, and project [Tungsten](https://databricks.com/blog/2015/04/28/project-tungsten-bringing-spark-closer-to-bare-metal.html). 

Dataset provides type-safety when working with SQL, since `Row` is mapped to a case class, so that each column can be referenced by property of that class.

> Note that performance for Dataset/DataFrames is the same across languages Scala, Java, Python, and R. This is due to the fact that the planning phase is just language-specific, only logical plan is constructed in Python, and all the physical execution is compiled and executed as JVM bytecode.

%md
You can also use different datasources that will be shown later or load Hive tables directly into Spark.

We have already created a table of social media usage from NYC 
> See the **Appendix** section below to create this `social_media_usage` table from raw data.

First let's make sure this table is available for us.

%md
As you can see the immutable value ``df`` is a DataFrame and more specifically it is:

> ``org.apache.spark.sql.DataFrame = [agency: string, platform: string, url: string, visits: int]``.

%md
Now let us print schema of the DataFrame ``df`` and have a look at the actual data:

%md
> Note that `(nullable = true)` simply means if the value is allowed to be `null`.

Let us count the number of rows in ``df``.

%md
So there are 5899 records or rows in the DataFrame `df`. Pretty good! You can also select individual columns using so-called DataFrame API, as follows:

%md
You can also apply ``.distinct()`` to extract only unique entries as follows:

%md
Let's see all the rows of the DataFrame `uniquePlatforms`. 

> Note that `display(uniquePlatforms)` unlike `uniquePlatforms.show()` displays all rows of the DataFrame + gives you ability to select different view, e.g. charts.

%md
### Spark SQL and DataFrame API
Spark SQL provides DataFrame API that can perform relational operations on both external data sources and internal collections, which is similar to widely used data frame concept in R, but evaluates operations support lazily (remember RDDs?), so that it can perform relational optimizations. This API is also available in Java, Python and R, but some functionality may not be available, although with every release of Spark people minimize this gap. 

So we give some examples how to query data in Python and R, but continue with Scala. You can do all DataFrame operations in this notebook using Python or R.

%md
Now it is time for some tips around how you use ``select`` and what the difference is between ``$"a"``, ``col("a")``, ``df("a")``. 

As you probably have noticed by now, you can specify individual columns to select by providing String values in select statement. But sometimes you need to:
- distinguish between columns with the same name
- use it to filter (actually you can still filter using full String expression)
- do some "magic" with joins and user-defined functions (this will be shown later)

So Spark gives you ability to actually specify columns when you select. Now the difference between all those three notations is ... none, those things are just aliases for a `Column` in Spark SQL, which means following expressions yield the same result:

```
// Using string expressions
df.select("agency", "visits")

// Using "$" alias for column
df.select($"agency", $"visits")

// Using "col" alias for column
df.select(col("agency"), col("visits"))

// Using DataFrame name for column
df.select(df("agency"), df("visits"))
```

This "same-difference" applies to filtering, i.e. you can either use full expression to filter, or column as shown in the following example:

```
// Using column to filter
df.select("visits").filter($"visits" > 100)

// Or you can use full expression as string
df.select("visits").filter("visits > 100")
```

> Note that ``$"visits" > 100`` expression looks amazing, but under the hood it is just another column, and it equals to ``df("visits").>(100)``, where, thanks to Scala paradigm ``>`` is just another function that you can define.

%md
Again you could have written the query above using any column aliases or String names or even writing the query directly. 

For example, we can do it using String names, as follows:

%md 
Refer to the [DataFrame API](http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.DataFrame) for more detailed API. In addition to simple column references and expressions, DataFrames also have a rich library of functions including string manipulation, date
arithmetic, common math operations and more. The complete list is available in the [DataFrame Function Reference](http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.functions$).

%md
Let's next explore some of the functionality that is available by transforming this DataFrame `df` into a new DataFrame called `fixedDF`.

* First, note that some columns are not exactly what we want them to be. 
  * For example date column should be standard Date/Timestamp SQL column, and 
  * visits should not contain null values, but `0`s instead. 
  
* Let us fix it using some code that is briefly explained here (don't worry if you don't get it completely now, you will get the hang of it by playing more)
    * The `coalesce` function is similar to `if-else` statement, i.e., if first column in expression is `null`, then the value of the second column is used and so on. 
    * `lit` just means column of constant value (`lit`erally speaking!).
    * the "funky" time conversion is essentially conversion from current format -> unix timestamp as a number -> Spark SQL Date format
    * we also remove `TOTAL` value from `platform` column.

%md
Okay, this is better, but `url`s are still inconsistent. 

Let's fix this by writing our own UDF (user-defined function) to deal with special cases. 

Note that if you **CAN USE Spark functions library**, do it. But for the sake of the example, custom UDF is shown below.

We take value of a column as String type and return the same String type, but ignore values that do not start with `http`.

%md
Let us apply our UDF on `fixedDF` to create a new DataFrame called `cleanedDF` as follows:

%md
Now, let's check that it actually worked by seeing the first 5 rows of the `cleanedDF` whose `url` `isNull` and `isNotNull`, as follows:

%md
Now there is a suggestion from you manager's manager's manager that due to some perceived privacy concerns we want to replace `agency` with some unique identifier.

So we need to do the following:

* create unique list of agencies with ids and 
* join them with main DataFrame. 

Sounds easy, right? Let's do it.

%md
Those who want to understand left/right inner/outer joins can see the video lectures in Module 3 of Anthony Joseph's Introduction to Big data edX course.

%md
Note, that we could have done aggregation using DataFrame API instead of Spark SQL.

%md
Alright, now let's see some *cool* operations with window functions. 

Our next task is to compute ``(daily visits / monthly average)`` for all platforms.