Apache Spark - Join and Aggregation

Q: What is the DataFrame equivalent of the SQL statement SELECT count(*) AS total
A: .agg(count("*").alias("total"))

Q: What is the DataFrame equivalent of the SQL statementSELECT firstName FROM PeopleDistinctNames INNER JOIN SSADistinctNames ON firstName = ssaFirstName
A: peopleDistinctNamesDF.join(ssaDistinctNamesDF, peopleDistinctNamesDF(col("firstName")) == col("ssaFirstName"))

multiple conditions for filter in spark data frames

# TODO carensDF = peopleDF.filter(col("firstName") == "Caren" & col("gender")== "F").limit(10) display(carensDF)

solution

carensDF = peopleDF.filter((col("firstName") == "Caren") & (col("gender") == "F")).limit(10)

separate the 2 conditions in filter with brackets

or

carensDF = peopleDF.filter(col("firstName") == "Caren" ).filter(col("gender") == "F").limit(10)

Cannot convert column into bool: please use '&' for 'and', '|' for 'or', '~' for 'not' when building DataFrame boolean expressions.

--------------------------------------------------------------------------- ValueError Traceback (most recent call last) <command-3332536293827331> in <module>() 1 # TODO ----> 2 carensDF = peopleDF.filter(col("firstName") == "Caren" and col("gender")== "F").limit(10) 3 display(carensDF) /databricks/spark/python/pyspark/sql/column.py in __nonzero__(self) 680 681 def __nonzero__(self): --> 682 raise ValueError("Cannot convert column into bool: please use '&' for 'and', '|' for 'or', " 683 "'~' for 'not' when building DataFrame boolean expressions.") 684 __bool__ = __nonzero__ ValueError: Cannot convert column into bool: please use '&' for 'and', '|' for 'or', '~' for 'not' when building DataFrame boolean expressions.

Perform addition/Mulitplication/division in Apache Spark on a column

# TODO
peopleWithFixedSalaries20KDF = peopleWithFixedSalariesDF.filter(col("salary") >100000).limit(10).withColumn("salary10k",round(col("salary")/10000))
peopleWithFixedSalaries20KDF.show()


use

withColumn("NameOfNewColumn",col("nameOfExistingColumn")/col("nameOfExistingColumn"))

TypeError: unorderable types: str() < int()

Error when executing

# TODO
peopleWithFixedSalaries20KDF = peopleWithFixedSalariesDF.filter("salary" <20000).limit(10)

TypeError: unorderable types: str() < int()

--------------------------------------------------------------------------- TypeError Traceback (most recent call last) <command-3332536293827326> in <module>() 1 # TODO ----> 2 peopleWithFixedSalaries20KDF = peopleWithFixedSalariesDF.filter("salary" <20000).limit(10) TypeError: unorderable types: str() < int()

solution is

# TODO peopleWithFixedSalaries20KDF = peopleWithFixedSalariesDF.filter(col("salary") <20000).limit(10)

or

# TODO

peopleWithFixedSalaries20KDF = peopleWithFixedSalariesDF.filter("salary <20000").limit(10)

instead.

Datasets

The Apache Spark Dataset API provides a type-safe, object-oriented programming interface. DataFrame is an alias for an untyped Dataset [Row]. Datasets provide compile-time type safety—which means that production applications can be checked for errors before they are run—and they allow direct operations over user-defined classes. The Dataset API also offers high-level domain-specific language operations like sum(), avg(), join(), select(), groupBy(), making the code a lot easier to express, read, and write.

In this tutorial module, you will learn how to:

• Create sample data
• Load sample data
• View a DataSet
• Process and visualize the Dataset

We also provide a sample notebook that you can import to access and run all of the code examples included in the module.

Create sample data

There two ways to create Datasets: dynamically and by reading from a JSON file using SparkSession. First, for primitive types in examples or demos, you can create Datasets within a Scala or Python notebook or in your sample Spark application. For example, here’s a way to create a Dataset of 100 integers in a notebook. We use the spark variable to create 100 integers as Dataset[Long].

Copy

// range of 100 numbers to create a Dataset.
val range100 = spark.range(100)
range100.collect()

Load sample data

The more common way is to read a data file from an external data source, such HDFS, blob storage, NoSQL, RDBMS, or local filesystem. Spark supports multiple formats: JSON, CSV, Text, Parquet, ORC, and so on. To read a JSON file, you also use the SparkSession variable spark.

The easiest way to start working with Datasets is to use an example Azure Databricks dataset available in the /databricks-datasetsfolder accessible within the Azure Databricks workspace.

Copy

val df = spark.read.json("/databricks-datasets/samples/people/people.json")

At the time of reading the JSON file, Spark does not know the structure of your data. That is, it doesn’t know how you want to organize your data into a typed-specific JVM object. It attempts to infer the schema from the JSON file and creates a DataFrame = Dataset[Row] of generic Row objects.

You can explicitly convert your DataFrame into a Dataset reflecting a Scala class object by defining a domain-specific Scala case class and converting the DataFrame into that type:

Copy

// First, define a case class that represents a type-specific Scala JVM Object
case class Person (name: String, age: Long)

// Read the JSON file, convert the DataFrames into a type-specific JVM Scala object
// Person. At this stage Spark, upon reading JSON, created a generic
// DataFrame = Dataset[Rows]. By explicitly converting DataFrame into Dataset
// results in a type-specific rows or collection of objects of type Person
val ds = spark.read.json("/databricks-datasets/samples/people/people.json").as[Person]

You can do something similar with IoT device state information captured in a JSON file: define a case class, read the JSON file, and convert the DataFrame = Dataset[DeviceIoTData].

There are two reasons to convert a DataFrame into a type-specific JVM object. First, after an explicit conversion, for all relational and query expressions using Dataset API, you get compile-type safety. For example, if you use a filter operation using the wrong data type, Spark detects mismatch types and issues a compile error rather an execution runtime error, so that you catch errors earlier. Second, the Dataset API provides high-order methods, which makes code much easier to read and develop. In the section Process and visualize the Dataset, notice how using Dataset typed objects makes the code easier to express and read.

As in the Person example, here create a case class that encapsulates the Scala object. To access the file that contains IoT data, load the file /databricks-datasets/iot/iot_devices.json.

Copy

// define a case class that represents the device data.
case class DeviceIoTData (
  battery_level: Long,
  c02_level: Long,
  cca2: String,
  cca3: String,
  cn: String,
  device_id: Long,
  device_name: String,
  humidity: Long,
  ip: String,
  latitude: Double,
  longitude: Double,
  scale: String,
  temp: Long,
  timestamp: Long
)

// read the JSON file and create the Dataset from the ``case class`` DeviceIoTData
// ds is now a collection of JVM Scala objects DeviceIoTData
val ds = spark.read.json("/databricks-datasets/iot/iot_devices.json").as[DeviceIoTData]

View the Dataset

To view the data in a tabular format instead of exporting it to a third-party tool, you can use the Azure Databricks display()command. Once you have loaded the JSON data and converted it into a Dataset for your type-specific collection of JVM objects, you can view them as you would view a DataFrame, by using either display() or standard Spark commands, such as take(), foreach(), and println() API calls.

Copy

// display the dataset table just read in from the JSON file
display(ds)

Copy

// Using the standard Spark commands, take() and foreach(), print the first
// 10 rows of the Datasets.
ds.take(10).foreach(println(_))
// Print first 10 rows of a dataset

Process and visualize the Dataset

A Dataset has transformations and actions. Most important are the high-level domain specific operations such as sum(), select(), avg(), join(), and union(). For more information, see the Scala Dataset API.

In this example, you can use filter(), map(), groupBy(), and avg(), all higher-level methods, to create new Datasets. What’s noteworthy is that you can access the attributes by their names as defined in the case class. That is, use the dot notation to access individual fields. As such, it makes code easy to read and write.

Copy

// filter out all devices whose temperature exceed 25 degrees and generate
// another Dataset with three fields that of interest and then display
// the mapped Dataset
val dsTemp = ds.filter(d => d.temp > 25).map(d => (d.temp, d.device_name, d.cca3)
display(dsTemp)

Copy

// Apply higher-level Dataset API methods such as groupBy() and avg().
// Filter temperatures > 25, along with their corresponding
// devices' humidity, compute averages, groupBy cca3 country codes,
// and display the results, using table and bar charts
val dsAvgTmp = ds.filter(d => {d.temp > 25}).map(d => (d.temp, d.humidity, d.cca3)).groupBy($"_3").avg()

// display averages as a table, grouped by the country
display(dsAvgTmp)

Copy

// Select individual fields using the Dataset method select()
// where battery_level is greater than 6. Note this high-level
// domain specific language API reads like a SQL query
display(ds.select($"battery_level", $"c02_level", $"device_name").where($"battery_level" > 6).sort($"c02_level"))

Here is an animated gif showing how quickly you can go from table to map to charts using Datasets and Azure Databricks display()command.

An additional benefit of using the Databricks display() command is that you can quickly view this data with a number of embedded visualizations. For example, in a new cell, you can issue SQL queries and click the map to see the data. But first you must save your dataset, ds, as a temporary table.

Copy

// registering your Dataset as a temporary table to which you can issue SQL queries
ds.createOrReplaceTempView("iot_device_data")

Having saved the Dataset of DeviceIoTData as a temporary table, you can issue SQL queries to it.

Copy

%sql select cca3, count (distinct device_id) as device_id from iot_device_data group by cca3 order by device_id desc limit 100

Write your first Apache Spark job

Write your first Apache Spark job

To write your first Apache Spark job, you add code to the cells of a Azure Databricks notebook. This example uses Python. For more information, you can also reference the Apache Spark Quick Start Guide.

This first command lists the contents of a folder in the Databricks File System:

Copy

# Take a look at the file system
display(dbutils.fs.ls("/databricks-datasets/samples/docs/"))

The next command uses spark, the SparkSession available in every notebook, to read the README.md text file and create a DataFrame named textFile:

Copy

textFile = spark.read.text("/databricks-datasets/samples/docs/README.md")

To count the lines of the text file, apply the count action to the DataFrame:

Copy

textFile.count()

One thing you may notice is that the second command, reading the text file, does not generate any output while the third command, performing the count, does. The reason for this is that the first command is a transformation while the second one is an action. Transformations are lazy and run only when an action is run. This allows Spark to optimize for performance (for example, run a filter prior to a join), instead of running commands serially. For a complete list of transformations and actions, refer to the Apache Spark Programming Guide: Transformations and Actions.

Apache Spark interfaces

Spark interfaces

There are three key Spark interfaces that you should know about.

Resilient Distributed Dataset (RDD)

Apache Spark’s first abstraction was the RDD. It is an interface to a sequence of data objects that consist of one or more types that are located across a collection of machines (a cluster). RDDs can be created in a variety of ways and are the “lowest level” API available. While this is the original data structure for Apache Spark, you should focus on the DataFrame API, which is a superset of the RDD functionality. The RDD API is available in the Java, Python, and Scala languages.

DataFrame

These are similar in concept to the DataFrame you may be familiar with in the pandas Python library and the R language. The DataFrame API is available in the Java, Python, R, and Scala languages.

 

Dataset

A combination of DataFrame and RDD. It provides the typed interface that is available in RDDs while providing the convenience of the DataFrame. The Dataset API is available in the Java and Scala languages.

In many scenarios, especially with the performance optimizations embedded in DataFrames and Datasets, it will not be necessary to work with RDDs. But it is important to understand the RDD abstraction because:

• The RDD is the underlying infrastructure that allows Spark to run so fast and provide data lineage.
• If you are diving into more advanced components of Spark, it may be necessary to use RDDs.
• The visualizations within the Spark UI reference RDDs.

When you develop Spark applications, you typically use DataFrames and Datasets.