ScaDaMaLe Course site and book

Introduction to Spark SQL

• This notebook explains the motivation behind Spark SQL, one of Spark's main libraries built on top of Spark Core.
• It introduces interactive SparkSQL queries and visualizations
• This notebook uses content from Databricks SparkSQL notebook and SparkSQL programming guide

Some core resources on Spark SQL

• READ: https://people.csail.mit.edu/matei/papers/2015/sigmodsparksql.pdf
• Bookmark: https://jaceklaskowski.gitbooks.io/mastering-spark-sql/content/spark-sql.html

Some other resources on SQL and Spark SQL

• https://en.wikipedia.org/wiki/SQL
• https://en.wikipedia.org/wiki/Apache_Hive
• http://www.infoq.com/articles/apache-spark-sql
• https://databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data-science.html
• https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html

Some of them are embedded below in-place for your convenience.

displayHTML(frameIt("https://en.wikipedia.org/wiki/Apache_Hive#HiveQL",175))

READ - https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html

This is an elaboration of the Apache Spark latest sql-progamming-guide.

Overview

Spark SQL is a Spark module for structured data processing. Unlike the basic Spark RDD API, the interfaces provided by Spark SQL provide Spark with more information about the structure of both the data and the computation being performed. Internally, Spark SQL uses this extra information to perform extra optimizations. There are several ways to interact with Spark SQL including SQL and the Dataset API. When computing a result the same execution engine is used, independent of which API/language you are using to express the computation. This unification means that developers can easily switch back and forth between different APIs based on which provides the most natural way to express a given transformation.

All of the examples on this page use sample data included in the Spark distribution and can be run in the spark-shell, pyspark shell, or sparkR shell.

Datasets and DataFrames

A Dataset is a distributed collection of data. Dataset is a new interface added in Spark 1.6 that provides the benefits of RDDs (strong typing, ability to use powerful lambda functions) with the benefits of Spark SQL’s optimized execution engine. A Dataset can be constructed from JVM objects and then manipulated using functional transformations (map, flatMap, filter, etc.). The Dataset API is available in Scala and Java. Python does not have the support for the Dataset API. But due to Python’s dynamic nature, many of the benefits of the Dataset API are already available (i.e. you can access the field of a row by name naturally row.columnName). The case for R is similar.

A DataFrame is a Dataset organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R/Python, but with richer optimizations under the hood. DataFrames can be constructed from a wide array of sources such as: structured data files, tables in Hive, external databases, or existing RDDs. The DataFrame API is available in Scala, Java, Python, and R. In Scala and Java, a DataFrame is represented by a Dataset of Rows. In the Scala API, DataFrame is simply a type alias of Dataset[Row]. While, in Java API, users need to use Dataset<Row> to represent a DataFrame.

Throughout this document, we will often refer to Scala/Java Datasets of Rows as DataFrames.

Getting Started in Spark

Starting Point: SparkSession

The entry point into all functionality in Spark is the SparkSession. To create a basic SparkSession in your scala Spark code, just use SparkSession.builder():

import org.apache.spark.sql.SparkSession

val spark = SparkSession
  .builder()
  .appName("Spark SQL basic example")
  .config("spark.some.config.option", "some-value")
  .getOrCreate()

// For implicit conversions like converting RDDs to DataFrames
import spark.implicits._

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

spark // ready-made Spark-Session

res2: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@cc46118

Creating DataFrames

With a SparkSession or SQLContext, applications can create DataFrame

• from an existing RDD,
• from a Hive table, or
• from various other data sources.

Just to recap:

• A DataFrame is a distributed collection of data organized into named columns (it is not strogly typed).
• You can think of it as being 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 for optimized storage.

Note that performance for DataFrames is the same across languages Scala, Java, Python, and R. This is due to the fact that the only planning phase is language-specific (logical + physical SQL plan), not the actual execution of the SQL plan.

DataFrame Basics

1. An empty DataFrame

2. DataFrame from a range

3. DataFrame from an RDD

4. DataFrame Operations (aka Untyped Dataset Operations)

5. Running SQL Queries Programmatically

6. Creating Datasets

1. Making an empty DataFrame

Spark has some of the pre-built methods to create simple DataFrames

• let us make an Empty DataFrame

val emptyDF = spark.emptyDataFrame // Ctrl+Enter to make an empty DataFrame

emptyDF: org.apache.spark.sql.DataFrame = []

Not really interesting, or is it?

You Try!

Uncomment the following cell, put your cursor after emptyDF. below and hit Tab to see what can be done with emptyDF.

//emptyDF.

2. Making a DataFrame from a range

Let us make a DataFrame next

• from a range of numbers, as follows:

val rangeDF = spark.range(0, 3).toDF() // Ctrl+Enter to make DataFrame with 0,1,2
// sc.parallelize(1 to 3).toDF() 

rangeDF: org.apache.spark.sql.DataFrame = [id: bigint]

Note that Spark automatically names column as id and casts integers to type bigint for big integer or Long.

In order to get a preview of data in DataFrame use show() as follows:

rangeDF.show() // Ctrl+Enter

+---+
| id|
+---+
|  0|
|  1|
|  2|
+---+

3. Making a DataFrame from an RDD

• Make an RDD
• Conver the RDD into a DataFrame using the defualt .toDF() method
• Conver the RDD into a DataFrame using the non-default .toDF(...) method
• Do it all in one line

Let's first make an RDD using the sc.parallelize method, transform it by a map and perform the collect action to display it, as follows:

val rdd1 = sc.parallelize(1 to 5).map(i => (i, i*2))
rdd1.collect() // Ctrl+Enter

rdd1: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[171] at map at command-2971213210277924:1
res7: Array[(Int, Int)] = Array((1,2), (2,4), (3,6), (4,8), (5,10))

Next, let us convert the RDD into DataFrame using the .toDF() method, as follows:

val df1 = rdd1.toDF() // Ctrl+Enter 

df1: org.apache.spark.sql.DataFrame = [_1: int, _2: int]

As it is clear, the DataFrame has columns named _1 and _2, each of type int. Let us see its content using the .show() method next.

df1.show() // Ctrl+Enter

+---+---+
| _1| _2|
+---+---+
|  1|  2|
|  2|  4|
|  3|  6|
|  4|  8|
|  5| 10|
+---+---+

Note that by default, i.e. without specifying any options as in toDF(), the column names are given by _1 and _2.

We can easily specify column names as follows:

val df1 = rdd1.toDF("once", "twice") // Ctrl+Enter
df1.show()

+----+-----+
|once|twice|
+----+-----+
|   1|    2|
|   2|    4|
|   3|    6|
|   4|    8|
|   5|   10|
+----+-----+

df1: org.apache.spark.sql.DataFrame = [once: int, twice: int]

Of course, we can do all of the above steps to make the DataFrame df1 in one line and then show it, as follows:

val df1 = sc.parallelize(1 to 5)
            .map(i => (i, i*2))
            .toDF("once", "twice") //Ctrl+enter
df1.show()

+----+-----+
|once|twice|
+----+-----+
|   1|    2|
|   2|    4|
|   3|    6|
|   4|    8|
|   5|   10|
+----+-----+

df1: org.apache.spark.sql.DataFrame = [once: int, twice: int]

4. DataFrame Operations (aka Untyped Dataset Operations)

DataFrames provide a domain-specific language for structured data manipulation in Scala, Java, Python and R.

As mentioned above, in Spark 2.0, DataFrames are just Dataset of Rows in Scala and Java API. These operations are also referred as “untyped transformations” in contrast to “typed transformations” come with strongly typed Scala/Java Datasets.

Here we include some basic examples of structured data processing using Datasets:

// This import is needed to use the $-notation
import spark.implicits._
// Print the schema in a tree format
df1.printSchema()

root
 |-- once: integer (nullable = false)
 |-- twice: integer (nullable = false)

import spark.implicits._

// Select only the "name" column
df1.select("once").show()

+----+
|once|
+----+
|   1|
|   2|
|   3|
|   4|
|   5|
+----+

// Select both columns, but increment the double column by 1
df1.select($"once", $"once" + 1).show()

+----+----------+
|once|(once + 1)|
+----+----------+
|   1|         2|
|   2|         3|
|   3|         4|
|   4|         5|
|   5|         6|
+----+----------+

// Select both columns, but increment the double column by 1 and rename it as "oncemore"
df1.select($"once", ($"once" * 1).as("oncemore")).show()

+----+--------+
|once|oncemore|
+----+--------+
|   1|       1|
|   2|       2|
|   3|       3|
|   4|       4|
|   5|       5|
+----+--------+

df1.filter($"once" > 2).show()

+----+-----+
|once|twice|
+----+-----+
|   3|    6|
|   4|    8|
|   5|   10|
+----+-----+

// Count the number of distinct singles -  a bit boring
df1.groupBy("once").count().show()

+----+-----+
|once|count|
+----+-----+
|   1|    1|
|   2|    1|
|   3|    1|
|   5|    1|
|   4|    1|
+----+-----+

Let's make a more interesting DataFrame for groupBy with repeated elements so that the count will be more than 1.

df1.show()

+----+-----+
|once|twice|
+----+-----+
|   1|    2|
|   2|    4|
|   3|    6|
|   4|    8|
|   5|   10|
+----+-----+

val df11 = sc.parallelize(3 to 5).map(i => (i, i*2)).toDF("once", "twice") // just make a small one
df11.show()

+----+-----+
|once|twice|
+----+-----+
|   3|    6|
|   4|    8|
|   5|   10|
+----+-----+

df11: org.apache.spark.sql.DataFrame = [once: int, twice: int]

val df111 = df1.union(df11) // let's take the unionAll of df1 and df11 into df111
df111.show() // df111 is obtained by simply appending the rows of df11 to df1

+----+-----+
|once|twice|
+----+-----+
|   1|    2|
|   2|    4|
|   3|    6|
|   4|    8|
|   5|   10|
|   3|    6|
|   4|    8|
|   5|   10|
+----+-----+

df111: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [once: int, twice: int]

// Count the number of distinct singles -  a bit less boring
df111.groupBy("once").count().show()

+----+-----+
|once|count|
+----+-----+
|   1|    1|
|   2|    1|
|   3|    2|
|   5|    2|
|   4|    2|
+----+-----+

For a complete list of the types of operations that can be performed on a Dataset refer to the API Documentation.

In addition to simple column references and expressions, Datasets 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.

You Try!

Uncomment the two lines in the next cell, and then fill in the ??? below to get a DataFrame df2 whose first two columns are the same as df1 and whose third column named triple has values that are three times the values in the first column.

//val df2 = sc.parallelize(1 to 5).map(i => (i, i*2, i????)).toDF("single", "double", "triple") // Ctrl+enter after editing ???
//df2.show()

5. Running SQL Queries Programmatically

The sql function on a SparkSession enables applications to run SQL queries programmatically and returns the result as a DataFrame.

df1

res27: org.apache.spark.sql.DataFrame = [once: int, twice: int]

// Register the DataFrame as a SQL temporary view
df1.createOrReplaceTempView("sdtable")

val sqlDF = spark.sql("SELECT * FROM sdtable")
sqlDF.show()

+----+-----+
|once|twice|
+----+-----+
|   1|    2|
|   2|    4|
|   3|    6|
|   4|    8|
|   5|   10|
+----+-----+

sqlDF: org.apache.spark.sql.DataFrame = [once: int, twice: int]

spark.sql("SELECT * FROM SDTable WHERE once>2").show()

+----+-----+
|once|twice|
+----+-----+
|   3|    6|
|   4|    8|
|   5|   10|
+----+-----+

SELECT * FROM SDTable WHERE once>2

5. Using SQL for interactively querying a table is very powerful!

Note -- comments are how you add comments in SQL cells beginning with %sql.

• You can run SQL select * statement to see all columns of the table, as follows:
  • This is equivalent to the above `display(diamondsDF)' with the DataFrame

-- Ctrl+Enter to select all columns of the table
select * from SDTable

-- Ctrl+Enter to select all columns of the table
-- note table names of registered tables are case-insensitive
select * from sdtable

Global Temporary View

Temporary views in Spark SQL are session-scoped and will disappear if the session that creates it terminates. If you want to have a temporary view that is shared among all sessions and keep alive until the Spark application terminates, you can create a global temporary view. Global temporary view is tied to a system preserved database global_temp, and we must use the qualified name to refer it, e.g. SELECT * FROM global_temp.view1. See http://spark.apache.org/docs/latest/sql-programming-guide.html#global-temporary-view for details.

6. Creating Datasets

---

Datasets are similar to RDDs, however, instead of using Java serialization or Kryo they use a specialized Encoder to serialize the objects for processing or transmitting over the network. While both encoders and standard serialization are responsible for turning an object into bytes, encoders are code generated dynamically and use a format that allows Spark to perform many operations like filtering, sorting and hashing without deserializing the bytes back into an object.

val rangeDS = spark.range(0, 3) // Ctrl+Enter to make DataSet with 0,1,2; Note we added '.toDF()' to this to create a DataFrame

rangeDS: org.apache.spark.sql.Dataset[Long] = [id: bigint]

rangeDS.show() // the column name 'id' is made by default here

+---+
| id|
+---+
|  0|
|  1|
|  2|
+---+

We can have more complicated objects in a DataSet too.

// Note: Case classes in Scala 2.10 can support only up to 22 fields. To work around this limit,
// you can use custom classes that implement the Product interface
case class Person(name: String, age: Long)

// Encoders are created for case classes
val caseClassDS = Seq(Person("Andy", 32), Person("Erik",44), Person("Anna", 15)).toDS()
caseClassDS.show()

+----+---+
|name|age|
+----+---+
|Andy| 32|
|Erik| 44|
|Anna| 15|
+----+---+

defined class Person
caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

// Encoders for most common types are automatically provided by importing spark.implicits._
val primitiveDS = Seq(1, 2, 3).toDS()
primitiveDS.map(_ + 1).collect() // Returns: Array(2, 3, 4)

primitiveDS: org.apache.spark.sql.Dataset[Int] = [value: int]
res36: Array[Int] = Array(2, 3, 4)

df1

res38: org.apache.spark.sql.DataFrame = [once: int, twice: int]

df1.show

+----+-----+
|once|twice|
+----+-----+
|   1|    2|
|   2|    4|
|   3|    6|
|   4|    8|
|   5|   10|
+----+-----+

// let's make a case class for our DF so we can convert it to Dataset
case class singleAndDoubleIntegers(once: Integer, twice: Integer)

defined class singleAndDoubleIntegers

val ds1 = df1.as[singleAndDoubleIntegers]

ds1: org.apache.spark.sql.Dataset[singleAndDoubleIntegers] = [once: int, twice: int]

ds1.show()

+----+-----+
|once|twice|
+----+-----+
|   1|    2|
|   2|    4|
|   3|    6|
|   4|    8|
|   5|   10|
+----+-----+

Go through the databricks Introductions Now

• https://docs.databricks.com/spark/latest/dataframes-datasets/introduction-to-dataframes-scala.html

Recommended Homework

This week's recommended homework is a deep dive into the SparkSQL programming guide.

ScaDaMaLe Course site and book

This and the next sequence of notebooks are an elaboration of the http://spark.apache.org/docs/latest/sql-programming-guide.html by Ivan Sadikov and Raazesh Sainudiin.

Spark Sql Programming Guide

• Overview
  • SQL
  • DataFrames
  • Datasets
• 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
• Data Sources
  • Generic Load/Save Functions
    • Manually Specifying Options
    • Run SQL on files directly
    • Save Modes
    • Saving to Persistent Tables
  • Parquet Files
    • Loading Data Programmatically
    • Partition Discovery
    • Schema Merging
    • Hive metastore Parquet table conversion
      • Hive/Parquet Schema Reconciliation
      • Metadata Refreshing
    • Configuration
  • JSON Datasets
  • Hive Tables
    • Interacting with Different Versions of Hive Metastore
  • JDBC To Other Databases
  • Troubleshooting
• Performance Tuning
  • Caching Data In Memory
  • Other Configuration Options
• Distributed SQL Engine
  • Running the Thrift JDBC/ODBC server
  • Running the Spark SQL CLI
• SQL Reference

What could one do with these notebooks?

One could read the Spark SQL Programming Guide that is embedded below and also linked above while going through the cells and doing the YouTrys in the following notebooks.

Why might one do it?

This homework/self-study will help you solve the assigned lab and theory exercises in the sequel, much faster by introducing you to some basic knowledge you need about Spark SQL.

NOTE on intra-iframe html navigation within a notebook:

• When navigating in the html-page embedded as an iframe, as in the cell below, you can:
  • click on a link in the displayed html page to see the content of the clicked link and
  • then right-click on the page and click on the arrow keys <- and -> to go back or forward.

Let's go through the programming guide in databricks now

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

Spark SQL, DataFrames and Datasets Guide

Spark SQL is a Spark module for structured data processing. Unlike the basic Spark RDD API, the interfaces provided by Spark SQL provide Spark with more information about the structure of both the data and the computation being performed. Internally, Spark SQL uses this extra information to perform extra optimizations. There are several ways to interact with Spark SQL including SQL and the Dataset API. When computing a result, the same execution engine is used, independent of which API/language you are using to express the computation. This unification means that developers can easily switch back and forth between different APIs based on which provides the most natural way to express a given transformation.

All of the examples on this page use sample data included in the Spark distribution and can be run in the spark-shell, pyspark shell, or sparkR shell.

SQL

One use of Spark SQL is to execute SQL queries. Spark SQL can also be used to read data from an existing Hive installation. For more on how to configure this feature, please refer to the Hive Tables section. When running SQL from within another programming language the results will be returned as a Dataset/DataFrame. You can also interact with the SQL interface using the command-line or over JDBC/ODBC.

Datasets and DataFrames

A Dataset is a distributed collection of data. Dataset is a new interface added in Spark 1.6 that provides the benefits of RDDs (strong typing, ability to use powerful lambda functions) with the benefits of Spark SQL’s optimized execution engine. A Dataset can be constructed from JVM objects and then manipulated using functional transformations (map, flatMap, filter, etc.). The Dataset API is available in Scala and Java. Python does not have the support for the Dataset API. But due to Python’s dynamic nature, many of the benefits of the Dataset API are already available (i.e. you can access the field of a row by name naturally row.columnName). The case for R is similar.

A DataFrame is a Dataset organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R/Python, but with richer optimizations under the hood. DataFrames can be constructed from a wide array of sources such as: structured data files, tables in Hive, external databases, or existing RDDs. The DataFrame API is available in Scala, Java, Python, and R. In Scala and Java, a DataFrame is represented by a Dataset of Rows. In the Scala API, DataFrame is simply a type alias of Dataset[Row]. While, in Java API, users need to use Dataset<Row> to represent a DataFrame.

Throughout this document, we will often refer to Scala/Java Datasets of Rows as DataFrames.

Background and Preparation

• If you are unfamiliar with SQL please brush-up from the basic links below.
• SQL allows one to systematically explore any structured data (i.e., tables) using queries. This is necessary part of the data science process.

One can use the SQL Reference at https://spark.apache.org/docs/latest/sql-ref.html to learn SQL quickly.

displayHTML(frameIt("https://en.wikipedia.org/wiki/SQL",500))

displayHTML(frameIt("https://en.wikipedia.org/wiki/Apache_Hive#HiveQL",175))

displayHTML(frameIt("https://spark.apache.org/docs/latest/sql-ref.html",700))

ScaDaMaLe Course site and book

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

Getting Started

Spark Sql Programming Guide

• Starting Point: SparkSession
• Creating DataFrames
• Untyped Dataset Operations (aka DataFrame Operations)
• Running SQL Queries Programmatically
• Global Temporary View
• Creating Datasets
• Interoperating with RDDs
  • Inferring the Schema Using Reflection
  • Programmatically Specifying the Schema
• Scalar Functions
• Aggregate Functions

Getting Started

Starting Point: SparkSession

The entry point into all functionality in Spark is the SparkSession class and/or SQLContext/HiveContext. SparkSession is created for you as spark when you start spark-shell on command-line REPL or through a notebook server (databricks, zeppelin, jupyter, etc.). You will need to create SparkSession usually when building an application for submission to a Spark cluster. To create a basic SparkSession, just use SparkSession.builder():

import org.apache.spark.sql.SparkSession

val spark = SparkSession
  .builder()
  .appName("Spark SQL basic example")
  .config("spark.some.config.option", "some-value")
  .getOrCreate()

// For implicit conversions like converting RDDs to DataFrames
import spark.implicits._

Find full example code in the Spark repo at:

• https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala

SparkSession in Spark 2.0 provides builtin support for Hive features including the ability to write queries using HiveQL, access to Hive UDFs, and the ability to read data from Hive tables. To use these features, you do not need to have an existing Hive setup.

// You could get SparkContext and SQLContext from SparkSession
val sc = spark.sparkContext
val sqlContext = spark.sqlContext

But in Databricks notebook (similar to spark-shell) SparkSession is already created for you and is available as spark (similarly, sc and sqlContext are also available).

// Evaluation of the cell by Ctrl+Enter will print spark session available in notebook
spark

res0: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@7f57ef36

After evaluation you should see something like this, i.e., a reference to the SparkSession you just created:

res0: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@5a289bf5

Creating DataFrames

With a SparkSessions, applications can create Dataset or DataFrame from an existing RDD, from a Hive table, or from various datasources.

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.

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.

As an example, the following creates a DataFrame based on the content of a JSON file:

val df = spark.read.json("examples/src/main/resources/people.json")

// Displays the content of the DataFrame to stdout
df.show()
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

Find full example code at - https://raw.githubusercontent.com/apache/spark/master/examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala in the Spark repo.

To be able to try this example in databricks we need to load the people.json file into dbfs. Let us do this programmatically next.

// you should not have to uncomment the block below as we already loaded data from 002_02_dbcCEdataLoader notebook
/*
// the following lines merely fetch the file from the URL and load it into the dbfs for us to try in databricks
// getLines from the file at the URL
val peopleJsonLinesFromURL = scala.io.Source.fromURL("https://raw.githubusercontent.com/apache/spark/master/examples/src/main/resources/people.json").getLines
// remove any pre-existing file at the dbfs location
dbutils.fs.rm("/datasets/sds/spark-examples/people.json",recurse=true)
// convert the lines fetched from the URL to a Seq, then make it a RDD of String and finally save it as textfile to dbfs
sc.parallelize(peopleJsonLinesFromURL.toSeq).saveAsTextFile("/datasets/sds/spark-examples/people.json")
*/
// read the text file we just saved or already loaded and see what it has
sc.textFile("/datasets/sds/spark-examples/people.json").collect.mkString("\n")

res1: String =
{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}

val df = spark.read.json("/datasets/sds/spark-examples/people.json")

df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

// you can also read into df like this - makes the same dataframe as the above call
val df = spark.read.format("json").load("/datasets/sds/spark-examples/people.json")

df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

df.show()

+----+-------+
| age|   name|
+----+-------+
|  30|   Andy|
|  19| Justin|
|null|Michael|
+----+-------+

Untyped Dataset Operations (aka DataFrame Operations)

DataFrames provide a domain-specific language for structured data manipulation in Scala, Java, Python and R.

As mentioned above, in Spark 2.0 or higher, DataFrames are just Dataset of Rows in Scala and Java API. These operations are also referred as “untyped transformations” in contrast to “typed transformations” come with strongly typed Scala/Java Datasets.

Here we include some basic examples of structured data processing using Datasets:

// This import is needed to use the $-notation
import spark.implicits._
// Print the schema in a tree format
df.printSchema()

root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)

import spark.implicits._

// Select only the "name" column
df.select("name").show()

+-------+
|   name|
+-------+
|   Andy|
| Justin|
|Michael|
+-------+

// Select everybody, but increment the age by 1
df.select($"name", $"age" + 1).show()

+-------+---------+
|   name|(age + 1)|
+-------+---------+
|   Andy|       31|
| Justin|       20|
|Michael|     null|
+-------+---------+

// Select people older than 21
df.filter($"age" > 21).show()

+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+

// Count people by age
df.groupBy("age").count().show()

+----+-----+
| age|count|
+----+-----+
|  19|    1|
|  30|    1|
|null|    1|
+----+-----+

Find full example code at - https://raw.githubusercontent.com/apache/spark/master/examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala in the Spark repo.

For a complete list of the types of operations that can be performed on a Dataset, refer to the API Documentation.

In addition to simple column references and expressions, Datasets 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.

Running SQL Queries Programmatically

The sql function on a SparkSession enables applications to run SQL queries programmatically and returns the result as a DataFrame.

// Register the DataFrame as a SQL temporary view
df.createOrReplaceTempView("people")

val sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()

+----+-------+
| age|   name|
+----+-------+
|  30|   Andy|
|  19| Justin|
|null|Michael|
+----+-------+

sqlDF: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

Global Temporary View

Temporary views in Spark SQL are session-scoped and will disappear if the session that creates it terminates. If you want to have a temporary view that is shared among all sessions and keep alive until the Spark application terminates, you can create a global temporary view. Global temporary view is tied to a system preserved database global_temp, and we must use the qualified name to refer it, e.g. SELECT * FROM global_temp.view1.

// Register the DataFrame as a global temporary view
df.createGlobalTempView("people")

// Global temporary view is tied to a system preserved database `global_temp`
spark.sql("SELECT * FROM global_temp.people").show()

+----+-------+
| age|   name|
+----+-------+
|  30|   Andy|
|  19| Justin|
|null|Michael|
+----+-------+

// Global temporary view is cross-session
spark.newSession().sql("SELECT * FROM global_temp.people").show()

+----+-------+
| age|   name|
+----+-------+
|  30|   Andy|
|  19| Justin|
|null|Michael|
+----+-------+

Creating Datasets

See https://spark.apache.org/docs/latest/sql-getting-started.html#creating-datasets

Datasets are similar to RDDs, however, instead of using Java serialization or Kryo they use a specialized Encoder to serialize the objects for processing or transmitting over the network. While both encoders and standard serialization are responsible for turning an object into bytes, encoders are code generated dynamically and use a format that allows Spark to perform many operations like filtering, sorting and hashing without deserializing the bytes back into an object.

case class Person(name: String, age: Long)

// Encoders are created for case classes
val caseClassDS = Seq(Person("Andy", 32)).toDS()
caseClassDS.show()

+----+---+
|name|age|
+----+---+
|Andy| 32|
+----+---+

defined class Person
caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

// Encoders for most common types are automatically provided by importing spark.implicits._
val primitiveDS = Seq(1, 2, 3).toDS()
primitiveDS.map(_ + 1).collect() // Returns: Array(2, 3, 4)

primitiveDS: org.apache.spark.sql.Dataset[Int] = [value: int]
res18: Array[Int] = Array(2, 3, 4)

// DataFrames can be converted to a Dataset by providing a class. Mapping will be done by name
val path = "/datasets/sds/spark-examples/people.json"
val peopleDS = spark.read.json(path).as[Person]
peopleDS.show()

+----+-------+
| age|   name|
+----+-------+
|  30|   Andy|
|  19| Justin|
|null|Michael|
+----+-------+

path: String = /datasets/sds/spark-examples/people.json
peopleDS: org.apache.spark.sql.Dataset[Person] = [age: bigint, name: string]

Dataset is not available directly in PySpark or SparkR.

Interoperating with RDDs

Spark SQL supports two different methods for converting existing RDDs into Datasets. 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 while writing your Spark application.

The second method for creating Datasets 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 Datasets 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 Seqs or Arrays. This RDD can be implicitly converted to a DataFrame and then be registered as a table. Tables can be used in subsequent SQL statements.

sc.textFile("/datasets/sds/spark-examples/people.txt").collect.mkString("\n")

res22: String =
Michael, 29
Andy, 30
Justin, 19

// For implicit conversions from RDDs to DataFrames
import spark.implicits._

// make a case class
case class Person(name: String, age: Long)

// Create an RDD of Person objects from a text file, convert it to a Dataframe
val peopleDF = spark.sparkContext
  .textFile("/datasets/sds/spark-examples/people.txt")
  .map(_.split(","))
  .map(attributes => Person(attributes(0), attributes(1).trim.toLong))
  .toDF()

import spark.implicits._
defined class Person
peopleDF: org.apache.spark.sql.DataFrame = [name: string, age: bigint]

peopleDF.show

+-------+---+
|   name|age|
+-------+---+
|Michael| 29|
|   Andy| 30|
| Justin| 19|
+-------+---+

// Register the DataFrame as a temporary view
peopleDF.createOrReplaceTempView("people")

// SQL statements can be run by using the sql methods provided by Spark
val teenagersDF = spark.sql("SELECT name, age FROM people WHERE age BETWEEN 13 AND 19")

teenagersDF: org.apache.spark.sql.DataFrame = [name: string, age: bigint]

teenagersDF.show()

+------+---+
|  name|age|
+------+---+
|Justin| 19|
+------+---+

// The columns of a row in the result can be accessed by field index
teenagersDF.map(teenager => "Name: " + teenager(0)).show()

+------------+
|       value|
+------------+
|Name: Justin|
+------------+

// or by field name
teenagersDF.map(teenager => "Name: " + teenager.getAs[String]("name")).show()

+------------+
|       value|
+------------+
|Name: Justin|
+------------+

// advanced ...
// No pre-defined encoders for Dataset[Map[K,V]], define explicitly
//implicit val mapEncoder = org.apache.spark.sql.Encoders.kryo[Map[String, Any]]

// Primitive types and case classes can be also be defined as follows
// import more classes here...
//implicit val stringIntMapEncoder: Encoder[Map[String, Any]] = ExpressionEncoder()

// row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]
//teenagersDF.map(teenager => teenager.getValuesMap[Any](List("name", "age"))).collect()

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 Rows from the original RDD;
2. Create the schema represented by a StructType matching the structure of Rows in the RDD created in Step 1.
3. Apply the schema to the RDD of Rows via createDataFrame method provided by SparkSession.

For example:

import org.apache.spark.sql.Row

import org.apache.spark.sql.types._

// Create an RDD
val peopleRDD = spark.sparkContext.textFile("/datasets/sds/spark-examples/people.txt")

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
peopleRDD: org.apache.spark.rdd.RDD[String] = /datasets/sds/spark-examples/people.txt MapPartitionsRDD[236] at textFile at command-2971213210278122:6

// The schema is encoded in a string
val schemaString = "name age"

schemaString: String = name age

// Generate the schema based on the string of schema
val fields = schemaString.split(" ")
  .map(fieldName => StructField(fieldName, StringType, nullable = true))

fields: Array[org.apache.spark.sql.types.StructField] = Array(StructField(name,StringType,true), StructField(age,StringType,true))

val schema = StructType(fields)

schema: org.apache.spark.sql.types.StructType = StructType(StructField(name,StringType,true),StructField(age,StringType,true))

// Convert records of the RDD (people) to Rows
val rowRDD = peopleRDD
  .map(_.split(","))
  .map(attributes => Row(attributes(0), attributes(1).trim))

rowRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[238] at map at command-2971213210278126:4

// Apply the schema to the RDD
val peopleDF = spark.createDataFrame(rowRDD, schema)

peopleDF: org.apache.spark.sql.DataFrame = [name: string, age: string]

peopleDF.show

+-------+---+
|   name|age|
+-------+---+
|Michael| 29|
|   Andy| 30|
| Justin| 19|
+-------+---+

// Creates a temporary view using the DataFrame
peopleDF.createOrReplaceTempView("people")

// SQL can be run over a temporary view created using DataFrames
val results = spark.sql("SELECT name FROM people")

results: org.apache.spark.sql.DataFrame = [name: string]

results.show

+-------+
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+

// The results of SQL queries are DataFrames and support all the normal RDD operations
// The columns of a row in the result can be accessed by field index or by field name
results.map(attributes => "Name: " + attributes(0)).show()

+-------------+
|        value|
+-------------+
|Name: Michael|
|   Name: Andy|
| Name: Justin|
+-------------+

Find full example code at - https://raw.githubusercontent.com/apache/spark/master/examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala in the Spark repo.

Scalar Functions

Scalar functions are functions that return a single value per row, as opposed to aggregation functions, which return a value for a group of rows. Spark SQL supports a variety of Built-in Scalar Functions. It also supports User Defined Scalar Functions.

Aggregate Functions

Aggregate functions are functions that return a single value on a group of rows. The Built-in Aggregation Functions provide common aggregations such as count(), countDistinct(), avg(), max(), min(), etc. Users are not limited to the predefined aggregate functions and can create their own. For more details about user defined aggregate functions, please refer to the documentation of User Defined Aggregate Functions.

ScaDaMaLe Course site and book

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

Getting Started - Exercise

After having gone through the simple example dataset in the programming guide, let's try a slightly larger dataset next.

Let us first create a table of social media usage from NYC

See the Load Data section 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 named 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.

// Let's find out what tables are already available for loading
spark.catalog.listTables.show(20,false) // only showing first 20 tables, if any

+----------------------------+--------+-----------+---------+-----------+
|name                        |database|description|tableType|isTemporary|
+----------------------------+--------+-----------+---------+-----------+
|all_prices                  |default |null       |MANAGED  |false      |
|bitcoin_normed_window       |default |null       |MANAGED  |false      |
|bitcoin_reversals_window    |default |null       |MANAGED  |false      |
|countrycodes                |default |null       |EXTERNAL |false      |
|gold_normed_window          |default |null       |MANAGED  |false      |
|gold_reversals_window       |default |null       |MANAGED  |false      |
|ltcar_locations_2_csv       |default |null       |MANAGED  |false      |
|magellan                    |default |null       |MANAGED  |false      |
|mobile_sample               |default |null       |EXTERNAL |false      |
|oil_normed_window           |default |null       |MANAGED  |false      |
|oil_reversals_window        |default |null       |MANAGED  |false      |
|oil_reversals_window2       |default |null       |MANAGED  |false      |
|over300all_2_txt            |default |null       |MANAGED  |false      |
|person                      |default |null       |MANAGED  |false      |
|personer                    |default |null       |MANAGED  |false      |
|persons                     |default |null       |MANAGED  |false      |
|simple_range                |default |null       |MANAGED  |false      |
|social_media_usage          |default |null       |MANAGED  |false      |
|social_media_usage_csv_gui  |default |null       |MANAGED  |false      |
|voronoi20191213uppsla1st_txt|default |null       |MANAGED  |false      |
+----------------------------+--------+-----------+---------+-----------+
only showing top 20 rows

NYC Social Media Usage Data

This dataset is from 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 could fetch this data, with slightly simplified column names, from the following URL:

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

Overview

Below we will show you how to create and query a table or DataFrame that you uploaded to DBFS. DBFS is a Databricks File System (their distributed file system) that allows you to store data for querying inside of Databricks.

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

Load Data

How to load csv file from URL and make a table in databricks

Okay, so how did we actually make table social_media_usage? Databricks allows us to upload/link external data and make it available as registerd SQL table. It involves several steps:

• load this social-media-usage.csv file from the following URL to databricks directly:
  • http://lamastex.org/datasets/public/NYCUSA/social-media-usage.csv
• Then load it into dbfs using the GUI and/or manipulate as follows programmatically.

// File location from URL is loaded
val socialMediaUsageFromURL = scala.io.Source.fromURL("http://lamastex.org/datasets/public/NYCUSA/social-media-usage.csv").getLines
// First line is the header - let's save it
val header = socialMediaUsageFromURL.next.split(",")
// remove any pre-existing file at the dbfs location
dbutils.fs.rm("/datasets/sds/social_media_usage.csv",recurse=true)
// convert the lines fetched from the URL to a Seq, then make it a RDD of String and finally save it as textfile to dbfs
sc.parallelize(socialMediaUsageFromURL.toSeq).saveAsTextFile("/datasets/sds/social_media_usage.csv")
// read the text file we just saved or already loaded and see what it has
sc.textFile("/datasets/sds/social_media_usage.csv").take(10).mkString("\n")

socialMediaUsageFromURL: Iterator[String] = <iterator>
header: Array[String] = Array(agency, platform, url, date, visits)
res1: String =
OEM,SMS,,2012-02-17,61652
OEM,SMS,,2012-11-09,44547
EDC,Flickr,http://www.flickr.com/nycedc,2012-05-09,
NYCHA,Newsletter,,2012-05-09,
DHS,Twitter,www.twitter.com/nycdhs,2012-06-13,389
DHS,Twitter,www.twitter.com/nycdhs,2012-08-02,431
DOH,Android,Condom Finder,2011-08-08,5026
DOT,Android,You The Man,2011-08-08,
MOME,Android,MiNY Venor app,2011-08-08,313
DOT,Broadcastr,,2011-08-08,

header

res2: Array[String] = Array(agency, platform, url, date, visits)

val socialMediaDF = spark
                      .read
                      .options(Map("infer_schema" -> "true"))
                      .csv("/datasets/sds/social_media_usage.csv")
                      .toDF(header:_*)

socialMediaDF: org.apache.spark.sql.DataFrame = [agency: string, platform: string ... 3 more fields]

socialMediaDF.count

res4: Long = 5898

// Let's create a view or table

val temp_table_name = "social_media_usage_temp"

socialMediaDF.createOrReplaceTempView(temp_table_name)

temp_table_name: String = social_media_usage_temp

// Let's find out what tables are already available for loading
spark.catalog.listTables.where($"name" startsWith "soc" ).show(5,false)

+--------------------------+--------+-----------+---------+-----------+
|name                      |database|description|tableType|isTemporary|
+--------------------------+--------+-----------+---------+-----------+
|social_media_usage        |default |null       |MANAGED  |false      |
|social_media_usage_csv_gui|default |null       |MANAGED  |false      |
|social_media_usage_temp   |null    |null       |TEMPORARY|true       |
+--------------------------+--------+-----------+---------+-----------+

With this registered as a temporary view, social_media_usage_temp will only be available to this particular notebook.

If you'd like other users to be able to query this table (in the databricks professional shard - not the free community edition; or in a managed on-premise cluster), you can also create a table from the DataFrame.

Once saved, this table will persist across cluster restarts as well as allow various users across different notebooks to query this data. To do so, choose your table name and use saveAsTable as done in the next cell.

val permanent_table_name = "social_media_usage"
socialMediaDF.write.mode("overwrite").format("parquet").saveAsTable(permanent_table_name)

permanent_table_name: String = social_media_usage

// Let's find out what tables starting with "soc" in their name are already available for loading
spark.catalog.listTables.where($"name" startsWith "soc" ).show(5,false)

+--------------------------+--------+-----------+---------+-----------+
|name                      |database|description|tableType|isTemporary|
+--------------------------+--------+-----------+---------+-----------+
|social_media_usage        |default |null       |MANAGED  |false      |
|social_media_usage_csv_gui|default |null       |MANAGED  |false      |
|social_media_usage_temp   |null    |null       |TEMPORARY|true       |
+--------------------------+--------+-----------+---------+-----------+

It looks like the table social_media_usage is available as a permanent table (isTemporary set as false), if you have not uncommented the last line in the previous cell (otherwise it will be available from a parquet file as a permanent table - we will see more about parquet in the sequel).

Next let us do the following:

• load this table as a DataFrame (yes, the dataframe already exists as socialMediaDF, but we want to make a new DataFrame directly from the table)
• print its schema and
• show the first 20 rows.

-- Ctrl+Enter to achieve the same result using standard SQL syntax!
SELECT * FROM social_media_usage

val df = spark.table("social_media_usage") // Ctrl+Enter

df: org.apache.spark.sql.DataFrame = [agency: string, platform: string ... 3 more fields]

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].

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

// Ctrl+Enter
df.printSchema() // prints schema of the DataFrame
df.show() // shows first n (default is 20) rows

root
 |-- agency: string (nullable = true)
 |-- platform: string (nullable = true)
 |-- url: string (nullable = true)
 |-- date: string (nullable = true)
 |-- visits: string (nullable = true)

+-----------------+----------+--------------------+----------+-------+
|           agency|  platform|                 url|      date| visits|
+-----------------+----------+--------------------+----------+-------+
|             MOAE|   YouTube|http://www.youtub...|2012-03-14|      5|
|          NYC Gov|   YouTube|http://www.youtub...|2012-03-14|      7|
|        NYC Water|   Youtube|http://www.youtub...|2012-03-14|   null|
|Veteran's Affairs|   Youtube|http://www.youtub...|2012-03-14|   null|
|             DRIS|   YouTube|http://www.youtub...|2012-03-14|   null|
|              TLC|  Facebook|http://www.facebo...|2012-03-14|    585|
|           nycgov|   Google+|https://plus.goog...|2012-03-14|   null|
|            TOTAL|     TOTAL|               TOTAL|2012-03-14|1688764|
|              DOH|   Android|       Condom Finder|2012-04-09|   null|
|              DOT|   Android|         You The Man|2012-04-09|   null|
|             MOME|   Android|      MiNY Venor app|2012-04-09|    343|
|              DOT|Broadcastr|                null|2012-04-09|   null|
|              DPR|Broadcastr|http://beta.broad...|2012-04-09|   null|
|            ENDHT|  Facebook|http://www.facebo...|2012-04-09|      9|
|              VAC|  Facebook|https://www.faceb...|2012-04-09|     55|
|           PlaNYC|  Facebook|http://www.facebo...|2012-04-09|     92|
|             DFTA|  Facebook|http://www.facebo...|2012-04-09|    178|
|              DOT|Broadcastr|                null|2012-10-24|   null|
|        energyNYC|  Facebook|http://www.facebo...|2012-04-09|    181|
|             MOIA|  Facebook|http://www.facebo...|2012-04-09|    171|
+-----------------+----------+--------------------+----------+-------+
only showing top 20 rows

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

Let us count the number of rows in df.

df.count() // Ctrl+Enter to get 5898

res11: Long = 5898

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:

val platforms = df.select("platform") // Shift+Enter

platforms: org.apache.spark.sql.DataFrame = [platform: string]

platforms.count() // Shift+Enter to count the number of rows

res13: Long = 5898

platforms.show(5) // Ctrl+Enter to show top 5 rows

+--------+
|platform|
+--------+
| YouTube|
| YouTube|
| Youtube|
| Youtube|
| YouTube|
+--------+
only showing top 5 rows

platforms.rdd.getNumPartitions

res15: Int = 2

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

val uniquePlatforms = df.select("platform").distinct() // Shift+Enter

uniquePlatforms: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [platform: string]

uniquePlatforms.count() // Ctrl+Enter to count the number of distinct platforms

res17: Long = 23

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.

uniquePlatforms.show(25,false)

+-------------------------+
|platform                 |
+-------------------------+
|nyc.gov                  |
|Flickr                   |
|Vimeo                    |
|iPhone                   |
|YouTube                  |
|WordPress                |
|SMS                      |
|iPhone App               |
|Youtube                  |
|Instagram                |
|iPhone app               |
|Linked-In                |
|Twitter                  |
|TOTAL                    |
|Tumblr                   |
|Newsletter               |
|Pinterest                |
|Broadcastr               |
|Android                  |
|Foursquare               |
|Google+                  |
|Foursquare (Badge Unlock)|
|Facebook                 |
+-------------------------+

//display(uniquePlatforms) // Ctrl+Enter to show all rows; use the scroll-bar on the right of the display to see all platforms

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.

# Ctrl+Enter to evaluate this python cell, recall '#' is the pre-comment character in python
# Using Python to query our "social_media_usage" table
pythonDF = spark.table("social_media_usage").select("platform").distinct()
pythonDF.show(3)

-- Ctrl+Enter to achieve the same result using standard SQL syntax!
select distinct platform from social_media_usage

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")

import org.apache.spark.sql.functions.col
// Using "col" alias for column
df.select(col("agency"), col("visits"))

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

import org.apache.spark.sql.functions.col
res20: org.apache.spark.sql.DataFrame = [agency: string, visits: string]

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")

res22: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [visits: string]

df.select("visits").filter('visits > 100).count // same as df.select("visits").filter($"visits" > 100)

res24: Long = 3279

df.select("visits").filter(col("visits") > 100).count// df.select("visits").filter($"visits" > 100)

res25: Long = 3279

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.

val sms = df.select($"agency", $"platform", $"visits").filter($"platform" === "SMS")
sms.show() // Ctrl+Enter

+------+--------+------+
|agency|platform|visits|
+------+--------+------+
|   DOE|     SMS|   382|
| NYCHA|     SMS|  null|
|   OEM|     SMS| 61652|
|   DOE|     SMS|   382|
| NYCHA|     SMS|  null|
|   OEM|     SMS| 61652|
|   DOE|     SMS|  null|
| NYCHA|     SMS|  null|
|   OEM|     SMS|  null|
|   DOE|     SMS|  1253|
| NYCHA|     SMS|  8300|
|   OEM|     SMS| 44547|
|   DOE|     SMS|  1253|
| NYCHA|     SMS|  8300|
|   OEM|     SMS| 44547|
|   DOE|     SMS|  1253|
| NYCHA|     SMS|  8300|
|   OEM|     SMS| 44547|
|   DOE|     SMS|  1253|
| NYCHA|     SMS|  8300|
+------+--------+------+
only showing top 20 rows

sms: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [agency: string, platform: string ... 1 more field]

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:

// Ctrl+Enter Note that we are using "platform = 'SMS'" since it will be evaluated as actual SQL
val sms = df.select(df("agency"), df("platform"), df("visits")).filter("platform = 'SMS'")
sms.show(5)

+------+--------+------+
|agency|platform|visits|
+------+--------+------+
|   DOE|     SMS|   382|
| NYCHA|     SMS|  null|
|   OEM|     SMS| 61652|
|   DOE|     SMS|   382|
| NYCHA|     SMS|  null|
+------+--------+------+
only showing top 5 rows

sms: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [agency: string, platform: string ... 1 more field]

Refer to the DataFrame API 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.

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 0s 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 (literally speaking!).
  • we also remove TOTAL value from platform column.

// Ctrl+Enter to make fixedDF

// import the needed sql functions
import org.apache.spark.sql.functions.{coalesce, lit}

// make the fixedDF DataFrame
val fixedDF = df.
   select(
     $"agency", 
     $"platform", 
     $"url", 
     $"date", 
     coalesce($"visits", lit(0)).as("visits"))
    .filter($"platform" =!= "TOTAL")

fixedDF.printSchema() // print its schema 
// and show the top 20 records fully
fixedDF.show(false) // the false argument does not truncate the rows, so you will not see something like this "anot..."

root
 |-- agency: string (nullable = true)
 |-- platform: string (nullable = true)
 |-- url: string (nullable = true)
 |-- date: string (nullable = true)
 |-- visits: string (nullable = false)

+-----------------+----------+--------------------------------------------------------------------------------------+----------+------+
|agency           |platform  |url                                                                                   |date      |visits|
+-----------------+----------+--------------------------------------------------------------------------------------+----------+------+
|MOAE             |YouTube   |http://www.youtube.com/YouCanTooNYC                                                   |2012-03-14|5     |
|NYC Gov          |YouTube   |http://www.youtube.com/nycgov                                                         |2012-03-14|7     |
|NYC Water        |Youtube   |http://www.youtube.com/nycwater                                                       |2012-03-14|0     |
|Veteran's Affairs|Youtube   |http://www.youtube.com/channel/UCi6IvOszIb3hHPMUsaNKyXA                               |2012-03-14|0     |
|DRIS             |YouTube   |http://www.youtube.com/nycdeptofrecords                                               |2012-03-14|0     |
|TLC              |Facebook  |http://www.facebook.com/pages/NYC-Taxi-and-Limousine-Commission/101679939900978?v=wall|2012-03-14|585   |
|nycgov           |Google+   |https://plus.google.com/u/0/b/104030911277642419611/104030911277642419611/posts/p/pub |2012-03-14|0     |
|DOH              |Android   |Condom Finder                                                                         |2012-04-09|0     |
|DOT              |Android   |You The Man                                                                           |2012-04-09|0     |
|MOME             |Android   |MiNY Venor app                                                                        |2012-04-09|343   |
|DOT              |Broadcastr|null                                                                                  |2012-04-09|0     |
|DPR              |Broadcastr|http://beta.broadcastr.com/Echo.html?audioId=670026-4001                              |2012-04-09|0     |
|ENDHT            |Facebook  |http://www.facebook.com/pages/NYC-Lets-End-Human-Trafficking/125730490795659?sk=wall  |2012-04-09|9     |
|VAC              |Facebook  |https://www.facebook.com/pages/NYC-Voter-Assistance-Commission/110226709012110        |2012-04-09|55    |
|PlaNYC           |Facebook  |http://www.facebook.com/pages/New-York-NY/PlaNYC/160454173971169?ref=ts               |2012-04-09|92    |
|DFTA             |Facebook  |http://www.facebook.com/pages/NYC-Department-for-the-Aging/109028655823590            |2012-04-09|178   |
|DOT              |Broadcastr|null                                                                                  |2012-10-24|0     |
|energyNYC        |Facebook  |http://www.facebook.com/EnergyNYC?sk=wall                                             |2012-04-09|181   |
|MOIA             |Facebook  |http://www.facebook.com/ihwnyc                                                        |2012-04-09|171   |
|City Store       |Facebook  |http://www.facebook.com/citystorenyc                                                  |2012-04-09|214   |
+-----------------+----------+--------------------------------------------------------------------------------------+----------+------+
only showing top 20 rows

import org.apache.spark.sql.functions.{coalesce, lit}
fixedDF: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [agency: string, platform: string ... 3 more fields]

Okay, this is better, but urls 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.

// Ctrl+Enter to evaluate this UDF which takes a input String called "value"
// and converts it into lower-case if it begins with http and otherwise leaves it as null, so we sort of remove non valid web-urls
val cleanUrl = udf((value: String) => if (value != null && value.startsWith("http")) value.toLowerCase() else null)

cleanUrl: org.apache.spark.sql.expressions.UserDefinedFunction = SparkUserDefinedFunction($Lambda$9812/558964277@3b5bdfcc,StringType,List(Some(class[value[0]: string])),Some(class[value[0]: string]),None,true,true)

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

// Ctrl+Enter
val cleanedDF = fixedDF.select($"agency", $"platform", cleanUrl($"url").as("url"), $"date", $"visits")

cleanedDF: org.apache.spark.sql.DataFrame = [agency: string, platform: string ... 3 more fields]

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

// Shift+Enter
cleanedDF.filter($"url".isNull).show(5)

+------+----------+----+----------+------+
|agency|  platform| url|      date|visits|
+------+----------+----+----------+------+
|   DOH|   Android|null|2012-04-09|     0|
|   DOT|   Android|null|2012-04-09|     0|
|  MOME|   Android|null|2012-04-09|   343|
|   DOT|Broadcastr|null|2012-04-09|     0|
|   DOT|Broadcastr|null|2012-10-24|     0|
+------+----------+----+----------+------+
only showing top 5 rows

// Ctrl+Enter
cleanedDF.filter($"url".isNotNull).show(5, false) // false in .show(5, false) shows rows untruncated

+-----------------+--------+-------------------------------------------------------+----------+------+
|agency           |platform|url                                                    |date      |visits|
+-----------------+--------+-------------------------------------------------------+----------+------+
|MOAE             |YouTube |http://www.youtube.com/youcantoonyc                    |2012-03-14|5     |
|NYC Gov          |YouTube |http://www.youtube.com/nycgov                          |2012-03-14|7     |
|NYC Water        |Youtube |http://www.youtube.com/nycwater                        |2012-03-14|0     |
|Veteran's Affairs|Youtube |http://www.youtube.com/channel/uci6ivoszib3hhpmusankyxa|2012-03-14|0     |
|DRIS             |YouTube |http://www.youtube.com/nycdeptofrecords                |2012-03-14|0     |
+-----------------+--------+-------------------------------------------------------+----------+------+
only showing top 5 rows

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.

// Crtl+Enter
// Import Spark SQL function that will give us unique id across all the records in this DataFrame
import org.apache.spark.sql.functions.monotonically_increasing_id

// We append column as SQL function that creates unique ids across all records in DataFrames 
val agencies = cleanedDF.select(cleanedDF("agency"))
                        .distinct()
                        .withColumn("id", monotonically_increasing_id())
agencies.show(5)

+--------------------+---+
|              agency| id|
+--------------------+---+
|              PlaNYC|  0|
|                 HIA|  1|
|NYC Digital: exte...|  2|
|           NYCGLOBAL|  3|
|              nycgov|  4|
+--------------------+---+
only showing top 5 rows

import org.apache.spark.sql.functions.monotonically_increasing_id
agencies: org.apache.spark.sql.DataFrame = [agency: string, id: bigint]

// Ctrl+Enter
// And join with the rest of the data, note how join condition is specified 
val anonym = cleanedDF.join(agencies, cleanedDF("agency") === agencies("agency"), "inner").select("id", "platform", "url", "date", "visits")

// We also cache DataFrame since it can be quite expensive to recompute join
anonym.cache()

// Display result
anonym.show(5)

+-------------+--------+--------------------+----------+------+
|           id|platform|                 url|      date|visits|
+-------------+--------+--------------------+----------+------+
| 841813590016| YouTube|http://www.youtub...|2012-03-14|     5|
| 901943132160| YouTube|http://www.youtub...|2012-03-14|     7|
| 575525617664| Youtube|http://www.youtub...|2012-03-14|     0|
|1657857376256| Youtube|http://www.youtub...|2012-03-14|     0|
| 171798691840| YouTube|http://www.youtub...|2012-03-14|     0|
+-------------+--------+--------------------+----------+------+
only showing top 5 rows

anonym: org.apache.spark.sql.DataFrame = [id: bigint, platform: string ... 3 more fields]

spark.catalog.listTables().show() // look at the available tables

+--------------------+--------+-----------+---------+-----------+
|                name|database|description|tableType|isTemporary|
+--------------------+--------+-----------+---------+-----------+
|          all_prices| default|       null|  MANAGED|      false|
|bitcoin_normed_wi...| default|       null|  MANAGED|      false|
|bitcoin_reversals...| default|       null|  MANAGED|      false|
|        countrycodes| default|       null| EXTERNAL|      false|
|  gold_normed_window| default|       null|  MANAGED|      false|
|gold_reversals_wi...| default|       null|  MANAGED|      false|
|ltcar_locations_2...| default|       null|  MANAGED|      false|
|            magellan| default|       null|  MANAGED|      false|
|       mobile_sample| default|       null| EXTERNAL|      false|
|   oil_normed_window| default|       null|  MANAGED|      false|
|oil_reversals_window| default|       null|  MANAGED|      false|
|oil_reversals_win...| default|       null|  MANAGED|      false|
|    over300all_2_txt| default|       null|  MANAGED|      false|
|              person| default|       null|  MANAGED|      false|
|            personer| default|       null|  MANAGED|      false|
|             persons| default|       null|  MANAGED|      false|
|        simple_range| default|       null|  MANAGED|      false|
|  social_media_usage| default|       null|  MANAGED|      false|
|social_media_usag...| default|       null|  MANAGED|      false|
|voronoi20191213up...| default|       null|  MANAGED|      false|
+--------------------+--------+-----------+---------+-----------+
only showing top 20 rows

-- to remove a TempTable if it exists already
drop table if exists anonym

// Register table for Spark SQL, we also import "month" function 
import org.apache.spark.sql.functions.month

anonym.createOrReplaceTempView("anonym")

import org.apache.spark.sql.functions.month

-- Interesting. Now let's do some aggregation. Display platform, month, visits
-- Date column allows us to extract month directly

select platform, month(date) as month, sum(visits) as visits from anonym group by platform, month(date)