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

%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. If you don't see `social_media_usage` as a `name`d table in the output of the next cell then we first need to ingest this dataset. Let's do it using the databricks' GUI for creating `Data` as done next.

%md

## NYC Social Media Usage Data

This dataset is from [https://datahub.io/JohnSnowLabs/nyc-social-media-usage#readme](https://datahub.io/JohnSnowLabs/nyc-social-media-usage#readme)

The Demographic Reports are produced by the Economic, Demographic and Statistical Research unit within the Countywide Service Integration and Planning Management (CSIPM) Division of the Fairfax County Department of Neighborhood and Community Services. Information produced by the Economic, Demographic and Statistical Research unit is used by every county department, board, authority and the Fairfax County Public Schools. In addition to the small area estimates and forecasts, state and federal data on Fairfax County are collected and summarized, and special studies and Quantitative research are conducted by the unit.

We are going to fetch this data, with slightly simplified column names, from the following URL:

- http://lamastex.org/datasets/public/NYCUSA/social-media-usage.csv

To turn the dataset into a registered table we will load it using the GUI as follows:

- Download it to your local machine / laptop and then use the 'Data' button on the left to upload it (we will try this method now).
  - This will put your data in the `Filestore` in databricks' distributed file system.

### Overview 

Below we will show you how to create and query a table or DataFrame that you uploaded to DBFS. [DBFS](https://docs.databricks.com/user-guide/dbfs-databricks-file-system.html) is a Databricks File System (their distributed file system) that allows you to store data for querying inside of Databricks. This notebook assumes that you have a file already inside of DBFS that you would like to read from.

In other setups, you can have the data in s3 (say in AWS) or in hdfs in your hadoop cluster, etc.

Alternatively, you can use `curl` or `wget` to download it to the local file system in `/databricks/driver` and then load it into `dbfs`, after this you can use read it via `spark` session into a dataframe and register it as a hive table. 

You can also get the data directly from here (but in this case you need to change the column names in the databricks Data upload GUI or programmatically to follow this notebook):

- http://datahub.io/JohnSnowLabs/nyc-social-media-usage

%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, date: timestamp, visits: integer]``.

%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. 
  * 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!).
    * 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.

%md 
## Interoperating with RDDs

Spark SQL supports two different methods for converting existing RDDs into DataFrames. The first method uses reflection to infer the schema of an RDD that contains specific types of objects. This reflection based approach leads to more concise code and works well when you already know the schema. 

The second method for creating DataFrames is through a programmatic interface that allows you to construct a schema and then apply it to an
existing RDD. While this method is more verbose, it allows you to construct DataFrames when the columns and their types are not known until runtime.

### Inferring the Schema Using Reflection

The Scala interface for Spark SQL supports automatically converting an RDD containing case classes to a DataFrame. The case class defines the
schema of the table. The names of the arguments to the case class are read using reflection and become the names of the columns. Case classes
can also be nested or contain complex types such as Sequences or Arrays. This RDD can be implicitly converted to a DataFrame and then be registered as a table.

%md
### Programmatically Specifying the Schema

When case classes cannot be defined ahead of time (for example, the structure of records is encoded in a string, or a text dataset will be parsed and fields will be projected differently for different users), a `DataFrame` can be created programmatically with three steps.

1.  Create an RDD of `Row`s from the original RDD
2.  Create the schema represented by a [StructType](https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.types.StructType) and [StructField](https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.types.StructField) classes matching the structure of `Row`s in the RDD created in Step 1.
3.  Apply the schema to the RDD of `Row`s via `createDataFrame` method provided by `SQLContext`.

%md
> **Side-note:** Dataset API is first-class citizen in Spark, and DataFrame is an alias for Dataset[Row]. Note that Python and R use DataFrames (since they are dynamically typed), but it is essentially a Dataset.