Spark’s Basic Data Types

Matching its supported programming languages, Spark supports basic internal data types. These data types can be declared in your Spark application or defined in your schema. For example, in Scala, you can define or declare a particular column name to be of type String or Byte or Long.

$SPARK_HOME/bin/spark-shell…
          scala> import org.apache.spark.sql.types._
          import org.apache.spark.sql.types._
          scala> val nameTypes = StringType
          nameTypes: org.apache.spark.sql.types.StringType.type = StringType
          scala> val firstName = nameTypes
          firstName: org.apache.spark.sql.types.StringType.type = StringType
          scala> val lastName = nameTypes
          lastName: org.apache.spark.sql.types.StringType.type = StringType

Here are the other basic data types supported in Spark.^{8 9} They all are subtypes of class DataTypes, except for DecimalType.

Table 3-2 Scala Basic Data Types in Spark

Likewise, Python too supports similar basic data types as enumerated in the table below: ^{10 11}

Table 3-3 Python Basic Data Types in Spark

Spark’s Structured and Complex Data Types

For complex data analytics, you will hardly deal only with simple or basic data types. Rather, your data will be complex, often structured or nested. For that you will need Spark to handle these complex data types. They come in many forms: maps, arrays, structs, date, timestamp, fields etc,.

Table 3-4 Scala Structured Data Types in Spark

The equivalent structured data types in Python are enumerated below:^{12 13}

Table 3-5 Python Structured Data Types in Spark

While these tables showcase the myriad types supported, it’s far more important to see how these types come together when you define a schema for your data.

Schemas and Creating DataFrames

A schema in Spark defines the column names and its associated data types for a DataFrame. More often schemas come into play when you are reading structured data from DataSources (more on this in the next chapter). Defining schema as opposed to schema-on-read offers three benefits: 1) you relieve Spark from the onus of inferring data types 2) you prevent Spark from creating a separate job just to read a large portion of your file to ascertain the schema, which for a very large data file, can be expensive and time consuming, and 3) detects errors early if data doesn’t match the schema.

So we encourage you to always define your schema upfront whenever you want to read a very large file from a data source. For a short illustration, and using Table 3-1 for our data, let’s define a schema and use that schema to create a DataFrame.

// In Scala
            import org.apache.spark.sql.types._
            val schema = StructType(Array(StructField("author", StringType, false),
             StructField("title", StringType, false),
             StructField("pages", IntegerType, false)))
            # In Python
            from pyspark.sql.types import *
            schema = StructType([StructField("author", StringType(), False),
             StructField("title", StringType(), False),
             StructField("pages", IntegerType(), False)])
            To define the same schema using DDL is much simpler.
            // In Scala
            val schema = "author STRING, title STRING, pages INT"
            # In Python
            schema = "author STRING, title STRING, pages INT"

# In Python 
            from pyspark.sql.types import *
            from pyspark.sql import SparkSession
            # define schema for our data using DDL 
            schema = "`Id` INT,`First` STRING,`Last` STRING,`Url` STRING,`Published` STRING,`Hits` INT,`Campaigns` ARRAY<STRING>"
            # create our static data
            data = [[1, "Jules", "Damji", "https://tinyurl.1", "1/4/2016", 4535, ["twitter", "LinkedIn"]],
             [2, "Brooke","Wenig","https://tinyurl.2", "5/5/2018", 8908, ["twitter", "LinkedIn"]],
             [3, "Denny", "Lee", "https://tinyurl.3","6/7/2019",7659, ["web", "twitter", "FB", "LinkedIn"]],
             [4, "Tathagata", "Das","https://tinyurl.4", "5/12/2018", 10568, ["twitter", "FB"]],
             [5, "Matei","Zaharia", "https://tinyurl.5", "5/14/2014", 40578, ["web", "twitter", "FB", "LinkedIn"]],
             [6, "Reynold", "Xin", "https://tinyurl.6", "3/2/2015", 25568, ["twitter", "LinkedIn"]]
             ]
            # main program
            if __name__ == "__main__":
             # create a SparkSession
             spark = (SparkSession
             .builder
             .appName("Example-3_6")
             .getOrCreate())
             # create a DataFrame using the schema defined above
             blogs_df = spark.createDataFrame(data, schema)
             # show the DataFrame; it should reflect our table above
             blogs_df.show()
             print()
             # print the schema used by Spark to process the DataFrame
             print(blogs_df.printSchema())

$ python Example-3_6.py
            …
            …
            +-------+---------+-------+-----------------+---------+-----+----------------------------+
            |Id|First |Last |Url |Published|Hits |Campaigns |
            +-------+---------+-------+-----------------+---------+-----+----------------------------+
            |1 |Jules |Damji |https://tinyurl.1|1/4/2016 |4535 |[twitter, LinkedIn] |
            |2 |Brooke |Wenig |https://tinyurl.2|5/5/2018 |8908 |[twitter, LinkedIn] |
            |3 |Denny |Lee |https://tinyurl.3|6/7/2019 |7659 |[web, twitter, FB, LinkedIn]|
            |4 |Tathagata|Das |https://tinyurl.4|5/12/2018|10568|[twitter, FB] |
            |5 |Matei |Zaharia|https://tinyurl.5|5/14/2014|40578|[web, twitter, FB, LinkedIn]|
            |6 |Reynold |Xin |https://tinyurl.6|3/2/2015 |25568|[twitter, LinkedIn] |
            +-------+---------+-------+-----------------+---------+-----+----------------------------+
            root
             |-- Id: integer (nullable = false)
             |-- First: string (nullable = false)
             |-- Last: string (nullable = false)
             |-- Url: string (nullable = false)
             |-- Published: string (nullable = false)
             |-- Hits: integer (nullable = false)
             |-- Campaigns: array (nullable = false)
             | |-- element: string (containsNull = false)

StructType(List(StructField(“Id”,IntegerType,false),
            StructField(“First”,StringType,false),
            StructField(“Last”,StringType,false),
            StructField(“Url”,StringType,false),
            StructField(“Published”,StringType,false),
            StructField(“Hits”,IntegerType,false),
            StructField(“Campaigns”,ArrayType(StringType,true),false)))

// In Scala
            package main.scala.chapter3
            import org.apache.spark.sql.SparkSession
            import org.apache.spark.sql.types._
            object Example3_7 {
             def main(args: Array[String]) {
             val spark = SparkSession
             .builder
             .appName("Example-3_7")
             .getOrCreate()
             if (args.length <= 0) {
             println("usage Example3_7 <file path to blogs.json>")
             System.exit(1)
             }
             // get the path to the JSON file
             val jsonFile = args(0)
             // define our schema programmatically
             val schema = StructType(Array(StructField("Id", IntegerType,false),
             StructField("First", StringType, false),
             StructField("Last", StringType, false),
             StructField("Url", StringType, false),
             StructField("Published", StringType, false),
             StructField("Hits", IntegerType, false),
             StructField("Campaigns", ArrayType(StringType), false)))
             // Create a DataFrame by reading from the JSON file 
             // with a predefined Schema
             val blogsDF = spark.read.schema(schema).json(jsonFile)
             print()
             // show the DataFrame schema as output
             blogsDF.show(false)
             // print the schemas
             print(blogsDF.printSchema)
             print(blogsDF.schema)
             }
            }

()+---+---------+-------+-----------------+---------+-----+----------------------------+
            |Id |First |Last |Url |Published|Hits |Campaigns |
            +---+---------+-------+-----------------+---------+-----+----------------------------+
            |1 |Jules |Damji |https://tinyurl.1|1/4/2016 |4535 |[twitter, LinkedIn] |
            |2 |Brooke |Wenig |https://tinyurl.2|5/5/2018 |8908 |[twitter, LinkedIn] |
            |3 |Denny |Lee |https://tinyurl.3|6/7/2019 |7659 |[web, twitter, FB, LinkedIn]|
            |4 |Tathagata|Das |https://tinyurl.4|5/12/2018|10568|[twitter, FB] |
            |5 |Matei |Zaharia|https://tinyurl.5|5/14/2014|40578|[web, twitter, FB, LinkedIn]|
            |6 |Reynold |Xin |https://tinyurl.6|3/2/2015 |25568|[twitter, LinkedIn] |
            +---+---------+-------+-----------------+---------+-----+----------------------------+
            root
             |-- Id: integer (nullable = true)
             |-- First: string (nullable = true)
             |-- Last: string (nullable = true)
             |-- Url: string (nullable = true)
             |-- Published: string (nullable = true)
             |-- Hits: integer (nullable = true)
             |-- Campaigns: array (nullable = true)
             | |-- element: string (containsNull = true)
            StructType(StructField(“Id”,IntegerType,true), 
             StructField(“First”,StringType,true), 
             StructField(“Last”,StringType,true), 
             StructField(“Url”,StringType,true),
             StructField(“Published”,StringType,true), 
             StructField(“Hits”,IntegerType,true),
             StructField(“Campaigns”,ArrayType(StringType,true),true))

Columns and Expressions

As mentioned before, named columns in DataFrames are conceptually similar to named columns in pandas or R DataFrames or an RDBMS table: They describe the type of field. On their values, you can perform some operations using relational or computational expressions; you can list all columns by their names. But more importantly, in Spark’s supported languages, columns are like objects with public methods. As such you can manipulate column values with expressions, or a series of mathematical computations.

You can also use expressions on columns of other Spark data types as well. For example, you create a simple expression using the expr(“columnName * 5”) or (expr(“columnName - 5”) > col(anothercolumnName)). Expr is part of the pyspark.sql.functions and org.apache.spark.sql.functions package. Like any other function in that package, expr is like a function taking arguments that Spark will parse as an expression and compute its result. We show some expr examples below.

All three languages Scala ¹⁴, Java¹⁵, and Python¹⁶ have public methods associated with columns.

Note: The Spark documentation refers to both Col and Column. Column is the name of the object, but you can also abbreviate it as col. More importantly, Spark needs to understand whether you are working with a DataFrame Column type, not a string. For brevity, we shall use col in the book.

// In Scala 
          import org.apache.spark.sql.functions.{col, expr}
          blogsDF.columns
          Array[String] = Array(Campaigns, First, Hits, Id, Last, Published, Url)
          // access a particular column
          blogsDF.col("Id")
          // use an expression to compute a value
          blogsDF.select(expr("Hits * 2")).show(2)
          // or use col to compute value
          blogsDF.select(col("Hits") * 2).show(2)
          +----------+
          |(Hits * 2)|
          +----------+
          | 9070|
          | 17816|
          +----------+
          // use expression to compute big hitters for blogs
          // this adds a new column Big Hitters based on the conditional expression
          blogsDF.withColumn("Big Hitters", (expr("Hits > 10000"))).show()
          +---+---------+-------+-----------------+---------+-----+--------------------+------------+
          | Id| First| Last| Url|Published| Hits| Campaigns|Big Hitters|
          +---+---------+-------+-----------------+---------+-----+--------------------+------------+
          | 1| Jules| Damji|https://tinyurl.1| 1/4/2016| 4535| [twitter, LinkedIn]| false|
          | 2| Brooke| Wenig|https://tinyurl.2| 5/5/2018| 8908| [twitter, LinkedIn]| false|
          | 3| Denny| Lee|https://tinyurl.3| 6/7/2019| 7659|[web, twitter, FB...| false|
          | 4|Tathagata| Das|https://tinyurl.4|5/12/2018|10568| [twitter, FB]| true|
          | 5| Matei|Zaharia|https://tinyurl.5|5/14/2014|40578|[web, twitter, FB...| true|
          | 6| Reynold| Xin|https://tinyurl.6| 3/2/2015|25568| [twitter, LinkedIn]| true|
          +---+---------+-------+-----------------+---------+-----+--------------------+------------+
          // use expression to concatenate three columns, create a new column
          // and show the newly created concatenated column
          blogsDF.withColumn("AuthorsId", (concat(expr("First"), expr("Last"), expr("Id")))).select(expr("AuthorsId")).show(n=4)
          +-------------+
          | AuthorsId|
          +-------------+
          | JulesDamji1|
          | BrookeWenig2|
          | DennyLee3|
          |TathagataDas4|
          +-------------+

// And these three statements return the same value, showing that

// expr are same as column method call
          blogsDF.select(expr("Hits")).show(2)
          blogsDF.select(col("Hits")).show(2)
          blogsDF.select("Hits").show(2)
          +-----+
          | Hits|
          +-----+
          | 4535|
          | 8908|
          +-----+
          // Sort by column "Id" in descending order
          blogsDF.sort(col("Id").desc).show()
          blogsDF.sort($"Id".desc).show()
          +--------------------+---------+-----+---+-------+---------+-----------------+
          | Campaigns| First| Hits| Id| Last|Published| Url|
          +--------------------+---------+-----+---+-------+---------+-----------------+
          | [twitter, LinkedIn]| Reynold|25568| 6| Xin| 3/2/2015|https://tinyurl.6|
          |[web, twitter, FB...| Matei|40578| 5|Zaharia|5/14/2014|https://tinyurl.5|
          | [twitter, FB]|Tathagata|10568| 4| Das|5/12/2018|https://tinyurl.4|
          |[web, twitter, FB...| Denny| 7659| 3| Lee| 6/7/2019|https://tinyurl.3|
          | [twitter, LinkedIn]| Brooke| 8908| 2| Wenig| 5/5/2018|https://tinyurl.2|
          | [twitter, LinkedIn]| Jules| 4535| 1| Damji| 1/4/2016|https://tinyurl.1|
          +--------------------+---------+-----+---+-------+---------+-----------------+

Both these expressions—blogs_df.sort(col(“Id”).desc) and blogsDF.sort($"Id”.desc)—above are identical. That is, they sort the DataFrame column name “Id” in a descending order: one uses an explicit function col(“Id”), while the other uses “$” before the name of the column, which is a function in Spark that converts it to column.

Note: We have only scratched the surface and employed a couple of methods on Column object. For a complete list of all public methods for Column object, we refer you to the Spark documentation.¹⁷

Column objects in a DataFrame can’t exist in isolation; each column is part of a row in a record; and all rows constitute an entire DataFrame, which as we will see later in the chapter is really a Dataset[Row] in Scala.

Rows

A row in Spark is a generic Row object, containing one or more columns. And each column may be of the same data type (Integer or String) or distinct with different types (Integer, String, Map, Array, etc.). Because Row is an object in Spark and an ordered collection of fields, you can instantiate a Row in each of Spark’s supported languages and access its fields by an index starting at 0. ^{18 19}

// In Scala
          import org.apache.spark.sql.Row
          // create a Row
          val blogRow = Row(6, "Reynold", "Xin", "https://tinyurl.6", 255568, "3/2/2015", Array("twitter", "LinkedIn"))
          // access using index for individual items
          blogRow(1)
          res62: Any = Reynold
          # In Python
          from pyspark.sql import Row
          blog_row = Row(6, "Reynold", "Xin", "https://tinyurl.6", 255568, "3/2/2015", ["twitter", "LinkedIn"])
          # access using index for individual items
          blog_rowr[1]
          'Reynold'
          Row objects can be used to create DataFrames if you need them for quick interactivity and exploration. 
          # In Python 
          from pyspark.sql import Row
          from pyspark.sql.type import *
          # using DDL String to define a schema
          schema = "`Author` STRING, `State` STRING"
          rows = [Row("Matei Zaharia", "CA"), Row("Reynold Xin", "CA")]
          authors_df = spark.createDataFrame(rows, schema)
          authors_df.show()
          // In Scala
          import org.apache.spark.sql.Row
          import org.apache.spark.sql.types._
          val rows = Seq(("Matei Zaharia", "CA"), ("Reynold Xin", "CA"))
          val authorsDF = rows.toDF(“Author”, “State”) 
          authorsDF.show()
          +-------------+-----+
          | Author|State|
          +-------------+-----+
          |Matei Zaharia| CA|
          | Reynold Xin| CA|
          +-------------+-----+

Even though we demonstrated a quick way to create DataFrames from Row objects, in practice, you will want to read them from a file as illustrated earlier. In most cases, because your files are going to be huge, defining a schema and using it is a quicker and more efficient way to create DataFrames.

After you have created a large distributed DataFrame, you are going to want to perform some common data operations on them. Let’s examine some of these Spark operations with high-level relational operators in the Structured APIs.

DataFrameReader and DataFrameWriter

Reading and writing is much simpler in Spark 2.x because of these high-level abstraction and contributions from the community to connect to myriad data sources, including common NoSQL stores, RDMS, streaming engines (Apache Kafka and Kinesis), etc.

To begin with a data source, let’s read a large CSV file containing San Francisco fire calls.²² As noted above, we will define a schema for this file and use DataFrameReader class and its methods to instruct Spark what to do. Because this file contains 28 columns and over 4,380,660 records, it’s more efficient to define a schema than have Spark infer it.

Hint: If you don’t want to specify the schema, Spark can infer schema from a sample at a lesser cost. For example you can use the sampleRatio option.

// In Scala
          val sampleDF = spark.read
           .option("samplingRatio",0.001)
           .option("header", "true")
           .csv("path_to_csv_file")

Note: The original data set has over 60 columns. We dropped a few unnecessary columns for this example. For example, some dates had null and or invalid values.

# In Python define a schema 
          from pyspark.sql.types import *
          # Programmatic way to define a schema 
          fire_schema = StructType([StructField('CallNumber', IntegerType(), True),
           StructField('UnitID', StringType(), True),
           StructField('IncidentNumber', IntegerType(), True),
           StructField('CallType', StringType(), True), 
           StructField('CallDate', StringType(), True), 
           StructField('WatchDate', StringType(), True),
           StructField('CallFinalDisposition', StringType(), True),
           StructField('AvailableDtTm', StringType(), True),
           StructField('Address', StringType(), True), 
           StructField('City', StringType(), True), 
           StructField('Zipcode', IntegerType(), True), 
           StructField('Battalion', StringType(), True), 
           StructField('StationArea', StringType(), True), 
           StructField('Box', StringType(), True), 
           StructField('OriginalPriority', StringType(), True), 
           StructField('Priority', StringType(), True), 
           StructField('FinalPriority', IntegerType(), True), 
           StructField('ALSUnit', BooleanType(), True), 
           StructField('CallTypeGroup', StringType(), True),
           StructField('NumAlarms', IntegerType(), True),
           StructField('UnitType', StringType(), True),
           StructField('UnitSequenceInCallDispatch', IntegerType(), True),
           StructField('FirePreventionDistrict', StringType(), True),
           StructField('SupervisorDistrict', StringType(), True),
           StructField('Neighborhood', StringType(), True),
           StructField('Location', StringType(), True),
           StructField('RowID', StringType(), True),
           StructField("Delay", FloatType(), True)])
          # read the file using DataFrameReader using format CSV
          sf_fire_file = "databricks-datasets/learning-spark-v2/sf-fire/sf-fire-calls.csv"
          fire_df = spark.read.csv(sf_fire_file, header=True, schema=fire_schema)
          // In Scala it would be similar
          val fileSchema = StructType(Array(StructField("CallNumber", IntegerType, true),
           StructField("UnitID", StringType, true),
           StructField("IncidentNumber", IntegerType, true),
           StructField("CallType", StringType, true), 
           StructField("Location", StringType, true)),
           …
           …
           StructField("Location", StringType, true)),
           StructField("Delay", FloatType, true)))
          // read the file using the CSV DataFrameReader
          val sfFireFile = "databricks-datasets/learning-spark-v2/sf-fire/sf-fire-calls.csv"
          val fireDF = spark.read.schema(fireSchema)
           .option("header", "true")
           .csv(sfFireFile)

The code spark.read.csv creates a DataFrameReader object to read a CSV file type DataSource and reads the file according to the schema supplied. With success, it returns a DataFrame of rows and named columns with their respective types, as dictated in the schema.

Using a DataFrameWriter, you can do one of two things. First, write the DataFrame as another format into an external data source. As with DataFrameReader, DataFrameWriter supports multiple data sources. ^{23 24} Parquet, a popular columnar format, is the default format and it uses compression to store the data. And second, if written as Parquet, it preserves the schema as part of the Parquet metadata. Subsequent reads back into DataFrame do not require to manually supply a schema.

Saving DataFrame as Parquet File Format and SQL Table

Another common data operation is to explore and transform data, and then persist the file as Parquet or save it as a SQL table. Persisting a transformed DataFrame is as easy as reading it. For example, to persist the above DataFrame both as a file and table immediately after reading is simple:

// In Scala to save as a parquet file or Table 
          val parquetPath = …
          fireDF.write
           .format("parquet")
           .save(parquetPath)

Alternatively, you can save it as a table, which registers metadata with the Hive metastore (we will cover SQL managed and unmanaged tables, metastores, and DataFrames in the next chapter).

// In Scala to save as a parquet file or Table 
          val parquetTable = … //name of the table
          fireDF.write
           .format("parquet")
           .saveAsTable(parquetTable)
          # In Python
          parquet_table = … # name of the table
          (fire_df.write
           .format("parquet")
           .saveAsTable(parquetTable))

Let’s walk through common data operations on DataFrames after you have read the data.

Transformations & Actions

Now that we have an entire distributed DataFrame composed of SF Fire department calls in memory, the first thing you as a data engineer or data scientist will want to do is examine your data: see what the columns look like. That is, are they of the correct data type? Do they need conversion to the right data type? Do they have null values etc.

In Chapter 2 (see section on Transformation, actions, and Lazy evaluations), you got a glimpse on some transformations and actions as high-level API and Domain Specific Language (DSL) operators on DataFrames. What can we find out from our San Francisco Fire calls?

Projections and Filters

A projection in the relational parlance is the ability to select all or individual rows matching a certain relational condition, by using filters. In Spark, projections equate to the select() method, while filter can be expressed as filter() or where() methods. Let’s use both these methods to examine parts of our SF Fire incidents.

# In Python
          few_fire_df = (fire_df.select("IncidentNumber", "AvailableDtTm", "CallType") 
           .where(col("CallType") != "Medical Incident"))
          few_fire_df.show(5, truncate=False)
          // In Scala
          val fewFireDF = fireDF.select("IncidentNumber", "AvailableDtTm", "CallType")
           .where($"CallType" =!= "Medical Incident")
          fewFireDF.show(5, false)
          This code gives us the following output:
          +--------------+----------------------+----------------+
          |IncidentNumber|AvailableDtTm |CallType |
          +--------------+----------------------+----------------+
          |2003234 |01/11/2002 01:58:43 AM|Medical Incident|
          |2003233 |01/11/2002 02:10:17 AM|Medical Incident|
          |2003235 |01/11/2002 01:47:00 AM|Structure Fire |
          |2003235 |01/11/2002 01:51:54 AM|Structure Fire |
          |2003235 |01/11/2002 01:47:00 AM|Structure Fire |
          +--------------+----------------------+----------------+
          (only showing top 5 rows)

However, what if you are only interested in distinct CallTypes as the causes of the fire calls, and you want to count the total. These simple and expressive queries do the job:

# In Python, return the count using the action distinctCount()
          fire_df.select("CallType").where(col("CallType") != "null").distinctCount()
          # filter for only distinct non-null CallTypes from all the rows
          fire_df.select("CallType").where(col("CallType") != "null").distinct().show(10, False)
          // In Scala, return the count using the action count()
          fireDF.select("CallType").where(col("CallType") =!= "null").distinct().count()
          // In Scala, filter for only distinct non-null CallTypes from all the rows
          fireDF.select("CallType").where($"CallType" =!= lit("null")).distinct().show(10, false)

This code returns the output of 32 distinct call types for the fire calls.

Out[20]: 32
          +-----------------------------------+
          |CallType |
          +-----------------------------------+
          |Elevator / Escalator Rescue |
          |Marine Fire |
          |Aircraft Emergency |
          |Confined Space / Structure Collapse|
          |Administrative |
          |Alarms |
          |Odor (Strange / Unknown) |
          |Lightning Strike (Investigation) |
          |Citizen Assist / Service Call |
          |HazMat |
          +-----------------------------------+
          (only showing top 10 rows)

Renaming, Adding or Dropping Columns, and Aggregating

Sometimes you wish to rename particular columns for reasons of style or conventions; other times for readability or brevity. Our original column names in the SF Fire calls file had spaces in them. For example, the column “IncidentNumber” was “Incident Number.” In fact, all columns had spaces in them, which can be problematic, especially when you wish to write or save a DataFrame as a Parquet file (which prohibits spaces in column names).

By specifying the desired column name in the schema StructField, as we did, we effectively changed the ultimate name in the returning DataFrame for all our column names.

Alternatively, you could selectively rename columns with DataFrame’s public withColumnRenamed() method. For instance, let’s change our Delay column to ResponseDelayedinMin and issue the count query again.

# In Python
          new_fire_df = fire_df.withColumnRenamed("Delay", "ResponseDelayedinMins")
          new_fire_df.select("ResponseDelayedinMins").where(col("ResponseDelayedinMins") > 5).show(5, False)
          // In Scala
          val newFireDF = fireDF.withColumnRenamed("Delay", "ResponseDelayedinMins")
          newFireDF.select("ResponseDelayedinMins").where($"ResponseDelayedinMins" > 5).show(5, false)

This gives us a new renamed column:

+---------------------+
          |ResponseDelayedinMins|
          +---------------------+
          |5.233333 |
          |6.9333334 |
          |6.116667 |
          |7.85 |
          |77.333336 |
          +---------------------+
          (only showing top 5 rows)

Note: Because DataFrame transformations are immutable, in both the above queries, when we rename a column, we get a new DataFrame while retaining the original with the old column name.

Modifying the contents of a column or its type is often a common data operation during exploration. In some cases, the data is raw or dirty, or its types are not amenable to supply as arguments to relational operators. For example, in our SF Fire calls data set, the columns “CallDate”, “WatchDate”, and “AlarmDtTm” are strings rather than either Unix timestamp or SQL Date, which Spark supports and Spark can easily manipulate during transformations or actions, especially during a date or time analysis of your data.

So how do we convert them into a more usable format? It’s quite simple with a set of high-level API methods.

spark.sql.function has a set of to/from date/timestamp functions such as to_timestamp() or to_date().

# In Python 
          fire_ts_df = (fire_df.withColumn("IncidentDate", to_timestamp(col("CallDate"), "MM/dd/yyyy")).drop("CallDate") 
           .withColumn("OnWatchDate", to_timestamp(col("WatchDate"), "MM/dd/yyyy")).drop("WatchDate") 
           .withColumn("AvailableDtTS", to_timestamp(col("AvailableDtTm"), "MM/dd/yyyy hh:mm:ss aa")).drop("AvailableDtTm"))
          # select the converted columns
          fire_ts_df.select("IncidentDate", "OnWatchDate", "AvailableDtTS").show(5, False)
          // In Scala
          val fireTsDF = fireDF.withColumn("IncidentDate", to_timestamp(col("CallDate"), "MM/dd/yyyy")).drop("CallDate")
           .withColumn("OnWatchDate", to_timestamp(col("WatchDate"), "MM/dd/yyyy")).drop("WatchDate") 
           .withColumn("AvailableDtTS", to_timestamp(col("AvailableDtTm"), "MM/dd/yyyy hh:mm:ss aa")).drop("AvailableDtTm") 
          // select the converted columns
          fireTsDF.select("IncidentDate", "OnWatchDate", "AvailableDtTS").show(5, false)

The query above packs quite a punch. A number of things are happening. Let’s unpack them:

Convert the existing column’s data type from string to a Spark supported timestamp.

Use the new format as specified in the format string “MM/dd/yyyy” or “MM/dd/yyyy hh:mm:ss aa” where appropriate.

After converting to new data type, drop() the old column, and append the new one specified in the first argument to withColumn() method.

Assign the new modified DataFrame to fire_ts_df

This query results in new columns:

+-------------------+-------------------+-------------------+
          |IncidentDate |OnWatchDate |AvailableDtTS |
          +-------------------+-------------------+-------------------+
          |2002-01-11 00:00:00|2002-01-10 00:00:00|2002-01-11 01:58:43|
          |2002-01-11 00:00:00|2002-01-10 00:00:00|2002-01-11 02:10:17|
          |2002-01-11 00:00:00|2002-01-10 00:00:00|2002-01-11 01:47:00|
          |2002-01-11 00:00:00|2002-01-10 00:00:00|2002-01-11 01:51:54|
          |2002-01-11 00:00:00|2002-01-10 00:00:00|2002-01-11 01:47:00|
          +-------------------+-------------------+-------------------+
          (only showing top 5 rows)

Now that we have modified the dates, we can query using spark.sql.functions like mon(), year(), day() etc., for instance, to explore our data further. Let’s calculate all the years data of fire calls in our data set. We can also ask how many calls were logged in the last seven days.

# In Python 
          fire_ts_df.select(year('IncidentDate')).distinct().orderBy(year('IncidentDate')).show()
          // In Scala
          fireTSDF.select(year($"IncidentDate")).distinct().orderBy(year($"IncidentDate")).show()

This query gives us all the years up to 2018 in this data set:

+------------------+
          |year(IncidentDate)|
          +------------------+
          | 2000|
          | 2001|
          | 2002|
          | 2003|
          | 2004|
          | 2005|
          | 2006|
          | 2007|
          | 2008|
          | 2009|
          | 2010|
          | 2011|
          | 2012|
          | 2013|
          | 2014|
          | 2015|
          | 2016|
          | 2017|
          | 2018|
          +------------------+

So far in this section, we have explored a number of concepts: DataFrameReaders and DataFrameWriters; defining a schema and using it in the DataFrame readers; saving a DataFrame as a Parquet file or table; projecting and filtering selected columns from an existing DataFrame; and modifying, renaming, and dropping some columns—all common data operations.

One final common operation is grouping data by values in a column, and aggregating the data in some way, like simply counting it. This pattern of grouping and counting is as common as projecting and filtering. Let’s have a go at it.

Aggregations

What if we want to ask the question what were the common types of fire calls? And another question: what ZIP codes accounted for most of the fire calls? All these are common patterns of data analysis and exploration.

A handful of transformations and actions on DataFrame such as groupBy(), orderBy() and count() offer the ability to aggregate by column names and then aggregate count across them. ²⁵

For larger DataFrames on which you plan to conduct frequent or repeated queries, you could benefit from caching a DataFrame. We will cover DataFrames caching strategies and their benefits in later chapters.

Caution: Also, for extremely large DataFrames, collect() is dangerous and resource-heavy (expensive), as it can cause Out of Memory (OOM) exceptions. Unlike count(), which returns a single number to the driver, collect() returns a collection of all the Row objects back to the driver. Instead, to take a peek at some Row records, you’re better off with take(n), where n will return only the first n Row objects of the DataFrame.

Let’s take our first question: what were the common types of fire calls?

# In Python
          fire_ts_df.select("CallType").where(col("CallType").isNotNull()).groupBy("CallType").count().orderBy("count", ascending=False).show(n=10, truncate=False)
          // In Scala 
          fireTSDF.select("CallType").where(col("CallType").isNotNull).groupBy("CallType").count().orderBy(desc("count")).show(10, false)

Both produce the following output:

+-------------------------------+-------+
          |CallType |count |
          +-------------------------------+-------+
          |Medical Incident |2843475|
          |Structure Fire |578998 |
          |Alarms |483518 |
          |Traffic Collision |175507 |
          |Citizen Assist / Service Call |65360 |
          |Other |56961 |
          |Outside Fire |51603 |
          |Vehicle Fire |20939 |
          |Water Rescue |20037 |
          |Gas Leak (Natural and LP Gases)|17284 |
          +-------------------------------+-------+

From the above, we can conclude that the most common call type is medical incident.

Other Common DataFrame Operations

Along with the above, DataFrames provide descriptive statistical methods like min(), max(), sum(), avg(), etc. Here are some examples showing how to compute them with our SF Fire calls data set.

Let’s compute the sum of alarms, the average response time, and the min and max response time to all fire calls in our dataset.

# In Python
          fire_ts_df.select(sum("NumAlarms"), avg("ResponseDelayedinMins"), min("ResponseDelayedinMins"), max("ResponseDelayedinMins")).show()
          // In Scala
          fireTSDF.select(sum("NumAlarms"), avg("ResponseDelayedinMins"), min("ResponseDelayedinMins"), max("ResponseDelayedinMins")).show()

Both queries generate the following output:

+--------------+--------------------------+--------------------------+--------------------------+
          |sum(NumAlarms)|avg(ResponseDelayedinMins)|min(ResponseDelayedinMins)|max(ResponseDelayedinMins)|
          +--------------+--------------------------+--------------------------+--------------------------+
          | 4403441| 3.902170335891614| 0.016666668| 1879.6167|
          +--------------+--------------------------+--------------------------+--------------------------+

For more advanced statistics, common with data science workloads, read the DataFrame documentation for methods like stat(), describe(), correlation(), covariance(), sampleBy(), approxQuantile(), frequentItems() etc.

As you can see, it’s easy to compose and chain expressive queries with DataFrames’ high-level API and DSL operators. We can’t imagine the opacity and readability of the code if we were to do the same with RDDs!

Typed Objects, Untyped Objects, and Generic Rows

In Spark’s supported languages, only Dataset exists in Java and Scala, whereas in Python and R, only DataFrames APIs make sense. There are a few reasons for this. First, these languages are not compile-time type-safe; types are dynamically inferred or assigned during execution, not during compile time. And second, typed-object in Scala or Java is a characteristic of a Java Virtual Machine (JVM); types to objects or variables are bound at compile time, not during execution time. Our Table 3-6 distills it in a nutshell.

Table 3-6 Typed and Untyped Objects in Spark

Row is a generic object type in Spark, holding a collection of mixed types that can be accessed using an index. Internally, Spark manipulates Row objects, converting them to equivalent Spark types covered in Tables 3.4 and 3.5. For example, an Int as one of your fields in a Row will be mapped or converted to IntegerType or IntegerType() respectively for Scala or Java and Python.

// In Scala
          import org.apache.spark.sql.Row 
          val row = Row(350, true, "Learning Spark 2E", null)
          # In Python
          From pyspark.sql import Row
          row = Row(350, True, "Learning Spark 2E", None)
          Returns into val 
          row: org.apache.spark.sql.Row = [350,true,Learning Spark 2E,null]

Using index into the Row object, you can access individual fields using Row’s public getter methods.

// In Scala
          row.getInt(0)
          row.getBoolean(1)
          row.getString(2)

Returning these values:

res23: Int = 350
          res24: Boolean = true
          res25: String = Learning Spark 2E
          # In Python
          row[0]
          row[1]
          row[2]

Returning values:

Out[13]: 350
          Out[14]: True
          Out[15]: 'Learning Spark 2E'

By contrast, typed objects are actual Java or Scala class objects in the JVM. Each element in a Dataset maps to a JVM object.

Creating Datasets

As with creating DataFrames from data sources, you have to know a schema. In other words, you need to know the data types so that you can define your schema. Although for JSON and CSV, you can infer the schema, for large datasets inferring schema is resource-intensive (expensive). The easiest way to specify the schema of the resulting Dataset in Scala is to use case classes, which can be introspected to determine its schema. In Java, JavaBean classes are used (we discuss JavaBeans and case classes in chapter 6).

Scala: Case Classes

When you wish to instantiate your own domain-specific object as a Dataset, you can do so by defining a case class in Scala. As an example, let’s look at a collection of internet of devices in a JSON file (We use this file in the end-to-end example below).²⁹

Our file has rows of JSON strings that look as follows:

{"device_id": 198164, "device_name": "sensor-pad-198164owomcJZ", "ip": "80.55.20.25", "cca2": "PL", "cca3": "POL", "cn": "Poland", "latitude": 53.080000, "longitude": 18.620000, "scale": "Celsius", "temp": 21, "humidity": 65, "battery_level": 8, "c02_level": 1408, "lcd": "red", "timestamp" :1458081226051}

To express each JSON entry as DeviceIoTData, a domain-specific object, define a Scala case class:

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,
               lcd: String, longitude: Double, scale:String, temp: Long, 
              timestamp: Long)

Once defined, we can use it to read our file and convert the returned Dataset[Row] into Dataset[DeviceIoTData]

// In Scala
              val ds = spark.read.json("/databricks-datasets/learning-spark-v2/iot-devices/iot_devices.json").as[DeviceIoTData]
              ds: org.apache.spark.sql.Dataset[DeviceIoTData] = [battery_level: bigint, c02_level: bigint ... 13 more fields]
              ds.show(5, false)

Datasets Operations

Just as you can perform transformations and actions on DataFrames, so can you on Datasets. Depending on the kind of operations, the results type will vary.

// In Scala
            Val filterTempDS = ds.filter(d => {d.temp > 30 && d.humidity > 70})
            filterTempDS: org.apache.spark.sql.Dataset[DeviceIoTData] = [battery_level: bigint, c02_level: bigint ... 13 more fields]
            filterTempDs.show(5, false)

Note: In the query above, we used a function as a argument to the Dataset method filter() because the method is overloaded with many signatures. This method usage filter(func: (T) ⇒ Boolean): Dataset[T] takes a lambda function: func: (T) ⇒ Boolean as its argument.

One thing is quite simple and clear from the above query: filter() method can take a lambda function in Scala. The argument to the lambda function is a JVM object of type DeviceIoTData. As such, we can access its individual data fields using the “.” notation, like you would in a Scala class or JavaBean.

Another thing to note is that with DataFrames, you express your filter() conditions as SQL-like DSL operations, which are language agnostic, as we saw above in our Fire calls example. In Dataset, these operations use language-native expressions as Scala or Java code.

Here’s another example that results into another smaller Dataset.

// In Scala
            case class DeviceTempByCountry(temp: Long, device_name: String, device_id: Long, cca3: String)
            val dsTemp = ds.filter(d => {d.temp > 25})
             .map(d => (d.temp, d.device_name, d.device_id, d.cca3))
             .toDF("temp","device_name","device_id","cca3")
             .as[DeviceTempByCountry]
            dsTemp.show(5, false)
            +----+---------------------+---------+----+
            |temp|device_name |device_id|cca3|
            +----+---------------------+---------+----+
            |34 |meter-gauge-1xbYRYcj |1 |USA |
            |28 |sensor-pad-4mzWkz |4 |USA |
            |27 |sensor-pad-6al7RTAobR|6 |USA |
            |27 |sensor-pad-8xUD6pzsQI|8 |JPN |
            |26 |sensor-pad-10BsywSYUF|10 |USA |
            +----+---------------------+---------+----+
            only showing top 5 rows

Or you can inspect only the first row of your Dataset.

val device = dsTemp.first()
            print(device)
            DeviceTempByCountry(34,meter-gauge-1xbYRYcj,1,USA)device: DeviceTempByCountry = DeviceTempByCountry(34,meter-gauge-1xbYRYcj,1,USA)

Alternatively, you could express the same query using column names and then cast to a Dataset[DeviceTempByCountry].

// In Scala
            val dsTemp2 = ds.select($"temp", $"device_name", $"device_id", $"device_id", $"cca3").where("temp > 25").as[DeviceTempByCountry]

Tip: Semantically, select() is like map() in the above queries. In that, both these queries select fields and generate equivalent results.

To recap, the operations—filter(), map(), groupBy(), select(), take(), etc—on Datasets are similar to the ones on DataFrames. In a way, Datasets are similar to RDDs in that they provide a similar interface to its aforementioned methods and compile-time safety but with a much easier to read and an object-oriented programming interface.

When using Datasets, the underlying Spark SQL engine does all the creation, conversion, serialization and deserialization of JVM objects, and off-Java heap memory management with the help of Dataset Encoders.^{30 31} (We will cover more about Datasets and memory management in chapter 6.)

So Then When to use RDDs?

Consider these scenarios for using RDDs:

you are using a third-party package that’s written in RDD;

you can forgo code optimization, efficient space utilization, and performance benefits available with DataFrames and Datasets; and

you want to precisely instruct Spark how-to-do a query

What’s more, you can seamlessly move between DataFrame or Dataset and RDDs at will—by a simple API method call df.rdd. (Note that moving back-and-forth between RDD and DataFrame has its cost and should be avoided unless necessary.) After all, DataFrames and Datasets are built on top of RDDs, and they get decomposed to compact RDDs code during whole-stage-code-generation (we discuss this in the next section).

Finally, in the above sections with DataFrames, Datasets, and SQL, we got an intuition in how structure enables developers to use easy and friendly APIs to compose expressive queries on structured data. In other words, you instruct Spark what-to-do, not how-to-do using high-level operations and Spark ascertains what’s the most efficient way to build a query and generate compact code for you.

This process of building the most efficient query and generating compact code is the job of Spark SQL engine. It’s the substrate upon which the above structured APIs are built. Let’s peek under the hood to understand Spark SQL engine.

Phase 1: Analysis

In all the above code examples, Spark SQL engine will generate an Abstract Syntax Tree (AST)³⁹ for the SQL or DataFrame query. In this phase, any columns or table names will be resolved by consulting an internal Catalog, a programmatic interface to Spark SQL that holds a list of names of columns, data types, functions, tables, databases, etc. After all successful resolutions, the query proceeds to the next phase.

Phase 2: Logical Optimizations

As Figure 3-4 shows, this phase comprises two internal stages. Applying a standard-rule based optimization, the Catalyst Optimizer will first construct a set of multiple plans and then, using cost-based optimizer (CBO), assign costs to each plan. ⁴⁰ These plans are laid out as operator trees (like Figure 3-5); they may include, for example, the process of constant folding, predicate pushdown, projection pruning, boolean expression simplification, etc. ⁴¹ (More on these concepts in chapter 8). For each of the trees, a computation cost is associated. This is the input into the physical plan.

Phase 3: Physical Planning

In the physical planning phase, Spark SQL picks a logical plan and generates one physical plan, using physical operators that match the Spark execution engine.

Phase 4: Code Generation

The final phase of query optimization involves generating efficient Java bytecode to run on each machine. Because Spark SQL can operate datasets loaded in memory datasets, Spark can use a state-of-the-art compiler technology for code generation to speed up execution. In other words, it acts as a compiler. Project Tungsten, which facilitates whole-stage code-generation, plays a role here.⁴²