SDS-2.2, Scalable Data Science

Archived YouTube video of this live unedited lab-lecture:

Diamonds ML Pipeline Workflow - DataFrame ETL and EDA Part

This is the Spark SQL parts that are focussed on extract-transform-Load (ETL) and exploratory-data-analysis (EDA) parts of an end-to-end example of a Machine Learning (ML) workflow.

Why are we using DataFrames? This is because of the Announcement in the Spark MLlib Main Guide for Spark 2.2 https://spark.apache.org/docs/latest/ml-guide.html that "DataFrame-based API is primary API".

This notebook is a scalarific break-down of the pythonic 'Diamonds ML Pipeline Workflow' from the Databricks Guide.

We will see this example again in the sequel

For this example, we analyze the Diamonds dataset from the R Datasets hosted on DBC.

Later on, we will use the DecisionTree algorithm to predict the price of a diamond from its characteristics.

Here is an outline of our pipeline:

• Step 1. Load data: Load data as DataFrame
• Step 2. Understand the data: Compute statistics and create visualizations to get a better understanding of the data.
• Step 3. Hold out data: Split the data randomly into training and test sets. We will not look at the test data until after learning.
• Step 4. On the training dataset:
  • Extract features: We will index categorical (String-valued) features so that DecisionTree can handle them.
  • Learn a model: Run DecisionTree to learn how to predict a diamond's price from a description of the diamond.
  • Tune the model: Tune the tree depth (complexity) using the training data. (This process is also called model selection.)
• Step 5. Evaluate the model: Now look at the test dataset. Compare the initial model with the tuned model to see the benefit of tuning parameters.
• Step 6. Understand the model: We will examine the learned model and results to gain further insight.

In this notebook, we will only cover Step 1 and Step 2. above. The other Steps will be revisited in the sequel.

Step 1. Load data as DataFrame

This section loads a dataset as a DataFrame and examines a few rows of it to understand the schema.

For more info, see the DB guide on importing data.

// We'll use the Diamonds dataset from the R datasets hosted on DBC.
val diamondsFilePath = "dbfs:/databricks-datasets/Rdatasets/data-001/csv/ggplot2/diamonds.csv"

diamondsFilePath: String = dbfs:/databricks-datasets/Rdatasets/data-001/csv/ggplot2/diamonds.csv

sc.textFile(diamondsFilePath).take(2) // looks like a csv file as it should

res2: Array[String] = Array("","carat","cut","color","clarity","depth","table","price","x","y","z", "1",0.23,"Ideal","E","SI2",61.5,55,326,3.95,3.98,2.43)

val diamondsRawDF = sqlContext.read    // we can use sqlContext instead of SparkSession for backwards compatibility to 1.x
    .format("com.databricks.spark.csv") // use spark.csv package
    .option("header", "true") // Use first line of all files as header
    .option("inferSchema", "true") // Automatically infer data types
    //.option("delimiter", ",") // Specify the delimiter as comma or ',' DEFAULT
    .load(diamondsFilePath)

diamondsRawDF: org.apache.spark.sql.DataFrame = [_c0: int, carat: double ... 9 more fields]

//There are 10 columns.  We will try to predict the price of diamonds, treating the other 9 columns as features.
diamondsRawDF.printSchema()

root
 |-- _c0: integer (nullable = true)
 |-- carat: double (nullable = true)
 |-- cut: string (nullable = true)
 |-- color: string (nullable = true)
 |-- clarity: string (nullable = true)
 |-- depth: double (nullable = true)
 |-- table: double (nullable = true)
 |-- price: integer (nullable = true)
 |-- x: double (nullable = true)
 |-- y: double (nullable = true)
 |-- z: double (nullable = true)

Note: (nullable = true) simply means if the value is allowed to be null.

Let us count the number of rows in diamondsDF.

diamondsRawDF.count() // Ctrl+Enter

res4: Long = 53940

So there are 53940 records or rows in the DataFrame.

Use the show(n) method to see the first n (default is 20) rows of the DataFrame, as folows:

diamondsRawDF.show(10)

+---+-----+---------+-----+-------+-----+-----+-----+----+----+----+
|_c0|carat|      cut|color|clarity|depth|table|price|   x|   y|   z|
+---+-----+---------+-----+-------+-----+-----+-----+----+----+----+
|  1| 0.23|    Ideal|    E|    SI2| 61.5| 55.0|  326|3.95|3.98|2.43|
|  2| 0.21|  Premium|    E|    SI1| 59.8| 61.0|  326|3.89|3.84|2.31|
|  3| 0.23|     Good|    E|    VS1| 56.9| 65.0|  327|4.05|4.07|2.31|
|  4| 0.29|  Premium|    I|    VS2| 62.4| 58.0|  334| 4.2|4.23|2.63|
|  5| 0.31|     Good|    J|    SI2| 63.3| 58.0|  335|4.34|4.35|2.75|
|  6| 0.24|Very Good|    J|   VVS2| 62.8| 57.0|  336|3.94|3.96|2.48|
|  7| 0.24|Very Good|    I|   VVS1| 62.3| 57.0|  336|3.95|3.98|2.47|
|  8| 0.26|Very Good|    H|    SI1| 61.9| 55.0|  337|4.07|4.11|2.53|
|  9| 0.22|     Fair|    E|    VS2| 65.1| 61.0|  337|3.87|3.78|2.49|
| 10| 0.23|Very Good|    H|    VS1| 59.4| 61.0|  338| 4.0|4.05|2.39|
+---+-----+---------+-----+-------+-----+-----+-----+----+----+----+
only showing top 10 rows

If you notice the schema of diamondsRawDF you will see that the automatic schema inference of SqlContext.read method has cast the values in the column price as integer.

To cleanup:

• let's recast the column price as double for downstream ML tasks later and
• let's also get rid of the first column of row indices.

import org.apache.spark.sql.types.DoubleType
//we will convert price column from int to double for being able to model, fit and predict in downstream ML task
val diamondsDF = diamondsRawDF.select($"carat", $"cut", $"color", $"clarity", $"depth", $"table",$"price".cast(DoubleType).as("price"), $"x", $"y", $"z")
diamondsDF.cache() // let's cache it for reuse
diamondsDF.printSchema // print schema

root
 |-- carat: double (nullable = true)
 |-- cut: string (nullable = true)
 |-- color: string (nullable = true)
 |-- clarity: string (nullable = true)
 |-- depth: double (nullable = true)
 |-- table: double (nullable = true)
 |-- price: double (nullable = true)
 |-- x: double (nullable = true)
 |-- y: double (nullable = true)
 |-- z: double (nullable = true)

import org.apache.spark.sql.types.DoubleType
diamondsDF: org.apache.spark.sql.DataFrame = [carat: double, cut: string ... 8 more fields]

diamondsDF.show(10,false) // notice that price column has Double values that end in '.0' now

+-----+---------+-----+-------+-----+-----+-----+----+----+----+
|carat|cut      |color|clarity|depth|table|price|x   |y   |z   |
+-----+---------+-----+-------+-----+-----+-----+----+----+----+
|0.23 |Ideal    |E    |SI2    |61.5 |55.0 |326.0|3.95|3.98|2.43|
|0.21 |Premium  |E    |SI1    |59.8 |61.0 |326.0|3.89|3.84|2.31|
|0.23 |Good     |E    |VS1    |56.9 |65.0 |327.0|4.05|4.07|2.31|
|0.29 |Premium  |I    |VS2    |62.4 |58.0 |334.0|4.2 |4.23|2.63|
|0.31 |Good     |J    |SI2    |63.3 |58.0 |335.0|4.34|4.35|2.75|
|0.24 |Very Good|J    |VVS2   |62.8 |57.0 |336.0|3.94|3.96|2.48|
|0.24 |Very Good|I    |VVS1   |62.3 |57.0 |336.0|3.95|3.98|2.47|
|0.26 |Very Good|H    |SI1    |61.9 |55.0 |337.0|4.07|4.11|2.53|
|0.22 |Fair     |E    |VS2    |65.1 |61.0 |337.0|3.87|3.78|2.49|
|0.23 |Very Good|H    |VS1    |59.4 |61.0 |338.0|4.0 |4.05|2.39|
+-----+---------+-----+-------+-----+-----+-----+----+----+----+
only showing top 10 rows

//View DataFrame in databricks
// note this 'display' is a databricks notebook specific command that is quite powerful for visual interaction with the data
// other notebooks like zeppelin have similar commands for interactive visualisation
display(diamondsDF)

Truncated to 30 rows

Step 2. Understand the data

Let's examine the data to get a better understanding of what is there. We only examine a couple of features (columns), but it gives an idea of the type of exploration you might do to understand a new dataset.

For more examples of using Databricks's visualization (even across languages) see https://docs.databricks.com/user-guide/visualizations/index.html NOW.

We can see that we have a mix of

• categorical features (cut, color, clarity) and
• continuous features (depth, x, y, z).

Let's first look at the categorical features.

You can also select one or more individual columns using so-called DataFrame API.

Let us select the column cut from diamondsDF and create a new DataFrame called cutsDF and then display it as follows:

val cutsDF = diamondsDF.select("cut") // Shift+Enter

cutsDF: org.apache.spark.sql.DataFrame = [cut: string]

cutsDF.show(10) // Ctrl+Enter

+---------+
|      cut|
+---------+
|    Ideal|
|  Premium|
|     Good|
|  Premium|
|     Good|
|Very Good|
|Very Good|
|Very Good|
|     Fair|
|Very Good|
+---------+
only showing top 10 rows

Let us use distinct to find the distinct types of cut's in the dataset.

// View distinct diamond cuts in dataset
val cutsDistinctDF = diamondsDF.select("cut").distinct()

cutsDistinctDF: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [cut: string]

cutsDistinctDF.show()

+---------+
|      cut|
+---------+
|  Premium|
|    Ideal|
|     Good|
|     Fair|
|Very Good|
+---------+

Clearly, there are just 5 kinds of cuts.

// View distinct diamond colors in dataset
val colorsDistinctDF = diamondsDF.select("color").distinct() //.collect()
colorsDistinctDF.show()

+-----+
|color|
+-----+
|    F|
|    E|
|    D|
|    J|
|    G|
|    I|
|    H|
+-----+

colorsDistinctDF: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [color: string]

// View distinct diamond clarities in dataset
val claritiesDistinctDF = diamondsDF.select("clarity").distinct() // .collect()
claritiesDistinctDF.show()

+-------+
|clarity|
+-------+
|   VVS2|
|    SI1|
|     IF|
|     I1|
|   VVS1|
|    VS2|
|    SI2|
|    VS1|
+-------+

claritiesDistinctDF: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [clarity: string]

We can examine the distribution of a particular feature by using display(),

You Try!

1. Click on the chart icon and Plot Options, and setting:

2. Value=<id>

3. Series groupings='cut'

4. and Aggregation=COUNT.

5. You can also try this using columns "color" and "clarity"

display(diamondsDF.select("cut"))

Truncated to 30 rows

// come on do the same for color NOW!

// and clarity too...

** You Try!**

Now play around with display of the entire DF and choosing what you want in the GUI as opposed to a .select(...) statement earlier.

For instance, the following display(diamondsDF) shows the counts of the colors by choosing in the Plot Options a bar-chart that is grouped with Series Grouping as color, values as <id> and Aggregation as COUNT. You can click on Plot Options to see these settings and can change them as you wish by dragging and dropping.

 display(diamondsDF)

Truncated to 30 rows

Now let's examine one of the continuous features as an example.

//Select: "Plot Options..." --> "Display type" --> "histogram plot" and choose to "Plot over all results" OTHERWISE you get the image from first 1000 rows only
display(diamondsDF.select("carat"))

Truncated to 30 rows

The above histogram of the diamonds' carat ratings shows that carats have a skewed distribution: Many diamonds are small, but there are a number of diamonds in the dataset which are much larger.

• Extremely skewed distributions can cause problems for some algorithms (e.g., Linear Regression).
• However, Decision Trees handle skewed distributions very naturally.

Note: When you call display to create a histogram like that above, it will plot using a subsample from the dataset (for efficiency), but you can plot using the full dataset by selecting "Plot over all results". For our dataset, the two plots can actually look very different due to the long-tailed distribution.

We will not examine the label distribution for now. It can be helpful to examine the label distribution, but it is best to do so only on the training set, not on the test set which we will hold out for evaluation. These will be seen in the sequel

You Try! Of course knock youself out visually exploring the dataset more...

display(diamondsDF.select("cut","carat"))

Truncated to 30 rows

Try scatter plot to see pairwise scatter plots of continuous features.

display(diamondsDF) //Ctrl+Enter

Truncated to 30 rows

Note that columns of type string are not in the scatter plot!

diamondsDF.printSchema // Ctrl+Enter

root
 |-- carat: double (nullable = true)
 |-- cut: string (nullable = true)
 |-- color: string (nullable = true)
 |-- clarity: string (nullable = true)
 |-- depth: double (nullable = true)
 |-- table: double (nullable = true)
 |-- price: double (nullable = true)
 |-- x: double (nullable = true)
 |-- y: double (nullable = true)
 |-- z: double (nullable = true)

Let us run through some basic inteactive SQL queries next

• HiveQL supports =, <, >, <=, >= and != operators. It also supports LIKE operator for fuzzy matching of Strings
• Enclose Strings in single quotes
• Multiple conditions can be combined using and and or
• Enclose conditions in () for precedence
• ...
• ...

Why do I need to learn interactive SQL queries?

Such queries in the widely known declarative SQL language can help us explore the data and thereby inform the modeling process!!!

Using DataFrame API, we can apply a filter after select to transform the DataFrame diamondsDF to the new DataFrame diamondsDColoredDF.

Below, $ is an alias for column.

Let as select the columns named carat, colour, price where color value is equal to D.

val diamondsDColoredDF = diamondsDF.select("carat", "color", "price").filter($"color" === "D") // Shift+Enter

diamondsDColoredDF: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [carat: double, color: string ... 1 more field]

diamondsDColoredDF.show(10) // Ctrl+Enter

+-----+-----+-----+
|carat|color|price|
+-----+-----+-----+
| 0.23|    D|357.0|
| 0.23|    D|402.0|
| 0.26|    D|403.0|
| 0.26|    D|403.0|
| 0.26|    D|403.0|
| 0.22|    D|404.0|
|  0.3|    D|552.0|
|  0.3|    D|552.0|
|  0.3|    D|552.0|
| 0.24|    D|553.0|
+-----+-----+-----+
only showing top 10 rows

As you can see all the colors are now 'D'. But to really confirm this we can do the following for fun:

diamondsDColoredDF.select("color").distinct().show

+-----+
|color|
+-----+
|    D|
+-----+

Let's try to do the same in SQL for those who know SQL from before.

First we need to see if the table is registerd (not just the DataFrame), and if not we ened to register our DataFrame as a temporary table.

sqlContext.tables.show() // Ctrl+Enter to see available tables

+--------+--------------------+-----------+
|database|           tableName|isTemporary|
+--------+--------------------+-----------+
| default|          cities_csv|      false|
| default|       cleaned_taxes|      false|
| default|commdettrumpclint...|      false|
| default|   donaldtrumptweets|      false|
| default|             linkage|      false|
| default|             nations|      false|
| default|           newmplist|      false|
| default|       ny_baby_names|      false|
| default|       nzmpsandparty|      false|
| default|    pos_neg_category|      false|
| default|                 rna|      false|
| default|                samh|      false|
| default|              table1|      false|
| default|          test_table|      false|
| default|             uscites|      false|
+--------+--------------------+-----------+

Looks like diamonds is already there (if not just execute the following cell).

diamondsDF.createOrReplaceTempView("diamonds")

sqlContext.tables.show() // Ctrl+Enter to see available tables

+--------+--------------------+-----------+
|database|           tableName|isTemporary|
+--------+--------------------+-----------+
| default|          cities_csv|      false|
| default|       cleaned_taxes|      false|
| default|commdettrumpclint...|      false|
| default|   donaldtrumptweets|      false|
| default|             linkage|      false|
| default|             nations|      false|
| default|           newmplist|      false|
| default|       ny_baby_names|      false|
| default|       nzmpsandparty|      false|
| default|    pos_neg_category|      false|
| default|                 rna|      false|
| default|                samh|      false|
| default|              table1|      false|
| default|          test_table|      false|
| default|             uscites|      false|
|        |            diamonds|       true|
+--------+--------------------+-----------+

-- Shift+Enter to do the same in SQL
select carat, color, price from diamonds where color='D'

Truncated to 30 rows

Alternatively, one could just write the SQL statement in scala to create a new DataFrame diamondsDColoredDF_FromTable from the table diamonds and display it, as follows:

val diamondsDColoredDF_FromTable = sqlContext.sql("select carat, color, price from diamonds where color='D'") // Shift+Enter

diamondsDColoredDF_FromTable: org.apache.spark.sql.DataFrame = [carat: double, color: string ... 1 more field]

// or if you like use upper case for SQL then this is equivalent
val diamondsDColoredDF_FromTable = sqlContext.sql("SELECT carat, color, price FROM diamonds WHERE color='D'") // Shift+Enter

diamondsDColoredDF_FromTable: org.apache.spark.sql.DataFrame = [carat: double, color: string ... 1 more field]

// from version 2.x onwards you can call from SparkSession, the pre-made spark in spark-shell or databricks notebook
val diamondsDColoredDF_FromTable = spark.sql("SELECT carat, color, price FROM diamonds WHERE color='D'") // Shift+Enter

diamondsDColoredDF_FromTable: org.apache.spark.sql.DataFrame = [carat: double, color: string ... 1 more field]

display(diamondsDColoredDF_FromTable) // Ctrl+Enter to see the same DF!

Truncated to 30 rows

// You can also use the familiar wildchard character '%' when matching Strings
display(spark.sql("SELECT * FROM diamonds WHERE clarity LIKE 'V%'"))

Truncated to 30 rows

// Combining conditions
display(spark.sql("SELECT * FROM diamonds WHERE clarity LIKE 'V%' AND price > 10000"))

Truncated to 30 rows

// selecting a subset of fields
display(spark.sql("SELECT carat, clarity, price FROM diamonds WHERE color = 'D'"))

Truncated to 30 rows

//renaming a field using as
display(spark.sql("SELECT carat AS carrot, clarity, price FROM diamonds"))

Truncated to 30 rows

//sorting
display(spark.sql("SELECT carat, clarity, price FROM diamonds ORDER BY price DESC"))

Truncated to 30 rows

diamondsDF.printSchema // since price is double in the DF that was turned into table we can rely on the descenting sort on doubles

root
 |-- carat: double (nullable = true)
 |-- cut: string (nullable = true)
 |-- color: string (nullable = true)
 |-- clarity: string (nullable = true)
 |-- depth: double (nullable = true)
 |-- table: double (nullable = true)
 |-- price: double (nullable = true)
 |-- x: double (nullable = true)
 |-- y: double (nullable = true)
 |-- z: double (nullable = true)

// sort by multiple fields
display(spark.sql("SELECT carat, clarity, price FROM diamonds ORDER BY carat ASC, price DESC"))

Truncated to 30 rows

// use this to type cast strings into Int when the table is loaded with string-valued columns
//display(spark.sql("select cast(carat as Int) as carat, clarity, cast(price as Int) as price from diamond order by carat asc, price desc"))

// sort by multiple fields and limit to first 5
// I prefer lowercase for SQL - and you can use either in this course - but in the field do what your Boss or your colleagues prefer :)
display(spark.sql("select carat, clarity, price from diamonds order by carat desc, price desc limit 5"))

//aggregate functions
display(spark.sql("select avg(price) as avgprice from diamonds"))

//average operator is doing an auto-type conversion from int to double
display(spark.sql("select avg(cast(price as Integer)) as avgprice from diamonds"))

//aggregate function and grouping
display(spark.sql("select color, avg(price) as avgprice from diamonds group by color"))

Why do we need to know these interactive SQL queries?

Such queries can help us explore the data and thereby inform the modeling process!!!

Of course, if you don't know SQL then don't worry, we will be doing these things in scala using DataFrames.

Finally, those who are planning to take the Spark Developer Exams online, then you can't escape from SQL questions there...

Further Preparation

For more on SQL syntax, check the SQL tutorial on W3Schools Note that HiveQL supports only a subset of operations supported by SQL

See databricks guide on tables NOW-ish.