Simple DataFrame transformations and SQL expressions

Simple DataFrame transformations allow us to do most of the standard things one can do when working a row at a time.² You can still do many of the same operations defined on RDDs, except using Spark SQL expressions instead of arbitrary functions. To illustrate this we will start by examining the different kinds of filter operations available on DataFrames.

DataFrame functions, like filter, accept Spark SQL expressions instead of lambdas. These expressions allow the optimizer to understand what the condition represents, and with filter, it can often be used to skip reading unnecessary records.

To get started, let’s look at a SQL expression to filter our data for unhappy pandas using our existing schema. The first step is looking up the column that contains this information. In our case it is happy, and for our DataFrame (called df) we access the column through the apply function (e.g., df("happy")). The filter expression requires the expression to return a boolean value, and if you wanted to select happy pandas, the entire expression could be retrieving the column value. However, since we want to find the unhappy pandas, we can check to see that happy isn’t true using the !== operator as shown in Example 3-18.

    pandaInfo.filter(pandaInfo("happy") !== true)

This illustrates how to access a specific column from a DataFrame. For accessing other structures inside of DataFrames, like nested structs, keyed maps, and array elements, use the same apply syntax. So, if the first element in the attributes array represent squishiness, and you only want very squishy pandas, you can access that element by writing df("attributes")(0) >= 0.5.

Our expressions need not be limited to a single column. You can compare multiple columns in our “filter” expression. Complex filters like that shown in Example 3-19 are more difficult to push down to the storage layer, so you may not see the same speedup over RDDs that you see with simpler filters.

    pandaInfo.filter(
      pandaInfo("happy").and(pandaInfo("attributes")(0) > pandaInfo("attributes")(1))
    )

Spark SQL’s DataFrame API has a very large set of operators available. You can use all of the standard mathematical operators on floating points, along with the standard logical and bitwise operations (prefix with bitwise to distinguish from logical). Columns use === and !== for equality to avoid conflict with Scala internals. For columns of strings, startsWith/endsWith, substr, like, and isNull are all available. The full set of operations is listed in org.apache.spark.sql.Column and covered in Table 3-3 and Table 3-4.

In addition to the operators directly specified on the column, an even larger set of functions on columns exists in org.apache.spark.sql.functions, some of which we cover in Tables 3-5, 3-6, and 3-7. For illustration, this example shows the values for each column at a specific row, but keep in mind that these functions are called on columns, not values.

Beyond simply filtering out data, you can also produce a DataFrame with new columns or updated values in old columns. Spark uses the same expression syntax we discussed for filter, except instead of having to include a condition (like testing for equality), the results are used as values in the new DataFrame. To see how you can use select on complex and regular data types, Example 3-20 uses the Spark SQL explode function to turn an input DataFrame of PandaPlaces into a DataFrame of just PandaInfo as well as computing the “squishness” to “hardness” ratio of each panda.

    val pandaInfo = pandaPlace.explode(pandaPlace("pandas")){
      case Row(pandas: Seq[Row]) =>
        pandas.map{
          case (Row(
            id: Long,
            zip: String,
            pt: String,
            happy: Boolean,
            attrs: Seq[Double])) =>
            RawPanda(id, zip, pt, happy, attrs.toArray)
        }}
    pandaInfo.select(
      (pandaInfo("attributes")(0) / pandaInfo("attributes")(1))
        .as("squishyness"))

While all of these operations are quite powerful, sometimes the logic you wish to express is more easily encoded with if/else semantics. Example 3-21 is a simple example of this, and it encodes the different types of panda as a numeric value.⁴ The when and otherwise functions can be chained together to create the same effect.

  /**
    * Encodes pandaType to Integer values instead of String values.
    *
    * @param pandaInfo the input DataFrame
    * @return Returns a DataFrame of pandaId and integer value for pandaType.
    */
  def encodePandaType(pandaInfo: DataFrame): DataFrame = {
    pandaInfo.select(pandaInfo("id"),
      (when(pandaInfo("pt") === "giant", 0).
      when(pandaInfo("pt") === "red", 1).
      otherwise(2)).as("encodedType")
    )
  }

Specialized DataFrame transformations for missing and noisy data

Spark SQL also provides special tools for handling missing, null, and invalid data. By using isNan or isNull along with filters, you can create conditions for the rows you want to keep. For example, if you have a number of different columns, perhaps with different levels of precision (some of which may be null), you can use coalesce(c1, c2, ..) to return the first nonnull column. Similarly, for numeric data, nanvl returns the first non-NaN value (e.g., nanvl(0/0, sqrt(-2), 3) results in 3). To simplify working with missing data, the na function on DataFrame gives us access to some common routines for handling missing data in DataFrameNaFunctions.

Beyond row-by-row transformations

Sometimes applying a row-by-row decision, as you can with filter, isn’t enough. Spark SQL also allows us to select the unique rows by calling dropDuplicates, but as with the similar operation on RDDs (distinct), this can require a shuffle, so is often much slower than filter. Unlike with RDDs, dropDuplicates can optionally drop rows based on only a subset of the columns, such as an ID field, as shown in Example 3-22.

    pandas.dropDuplicates(List("id"))

This leads nicely into our next section on aggregates and groupBy since often the most expensive component of each is the shuffle.

Aggregates and groupBy

Spark SQL has many powerful aggregates, and thanks to its optimizer it can be easy to combine many aggregates into one single action/query. Like with Pandas’ DataFrames, groupBy returns special objects on which we can ask for certain aggregations to be performed. In pre-2.0 versions of Spark, this was a generic GroupedData, but in versions 2.0 and beyond, DataFrames groupBy is the same as one Datasets.

Aggregations on Datasets have extra functionality, returning a GroupedDataset (in pre-2.0 versions of Spark) or a KeyValueGroupedDataset when grouped with an arbitrary function, and a RelationalGroupedDataset when grouped with a relational/Dataset DSl expression. The additional typed functionality is discussed in “Grouped Operations on Datasets”, and the common “untyped” DataFrame and Dataset groupBy functionality is explored here.

min, max, avg, and sum are all implemented as convenience functions directly on GroupedData, and more can be specified by providing the expressions to agg. Example 3-23 shows how to compute the maximum panda size by zip code. Once you specify the aggregates you want to compute, you can get the results back as a DataFrame.

  def maxPandaSizePerZip(pandas: DataFrame): DataFrame = {
    pandas.groupBy(pandas("zip")).max("pandaSize")
  }

While Example 3-23 computes the max on a per-key basis, these aggregates can also be applied over the entire DataFrame or all numeric columns in a DataFrame. This is often useful when trying to collect some summary statistics for the data with which you are working. In fact, there is a built-in describe transformation which does just that, although it can also be limited to certain columns, which is used in Example 3-24 and returns Example 3-25.

    // Compute the count, mean, stddev, min, max summary stats for all
    // of the numeric fields of the provided panda infos. non-numeric
    // fields (such as string (name) or array types) are skipped.
    val df = pandas.describe()
    // Collect the summary back locally
    println(df.collect())

Array([count,3,3], [mean,1.3333333333333333,5.0],
  [stddev,0.5773502691896258,4.358898943540674], [min,1,2], [max,2,10])

For computing multiple different aggregations, or more complex aggregations, you should use the agg API on the GroupedData instead of directly calling count, mean, or similar convenience functions. For the agg API, you either supply a list of aggregate expressions, a string representing the aggregates, or a map of column names to aggregate function names. Once we’ve called agg with the requested aggregates, we get back a regular DataFrame with the aggregated results. As with regular functions, they are listed in the org.apache.spark.sql.functions Scaladoc. Table 3-8 lists some common and useful aggregates. For our example results in these tables we will consider a DataFrame with the schema of name field (as a string) and age (as an integer), both nullable with values ({"ikea", null}, {"tube", 6}, {"real", 30}). Example 3-26 shows how to compute both the min and mean for the pandaSize column on our running panda example.

  def minMeanSizePerZip(pandas: DataFrame): DataFrame = {
    // Compute the min and mean
    pandas.groupBy(pandas("zip")).agg(
      min(pandas("pandaSize")), mean(pandas("pandaSize")))
  }

If the built-in aggregation functions don’t meet your needs, you can extend Spark SQL using UDFs as discussed in “Extending with User-Defined Functions and Aggregate Functions (UDFs, UDAFs)”, although things can be more complicated for aggregate functions.

Windowing

Spark SQL 1.4.0 introduced windowing functions to allow us to more easily work with ranges or windows of rows. When creating a window you specify what columns the window is over, the order of the rows within each partition/group, and the size of the window (e.g., K rows before and J rows after OR range between values). If it helps to think of this visually, Figure 3-2 shows a sample window and its results. Using this specification each input row is related to some set of rows, called a frame, that is used to compute the resulting aggregate. Window functions can be very useful for things like computing average speed with noisy data, relative sales, and more. A window for pandas by age is shown in Example 3-27.

    val windowSpec = Window
      .orderBy(pandas("age"))
      .partitionBy(pandas("zip"))
      .rowsBetween(start = -10, end = 10) // can use rangeBetween for range instead

Figure 3-2. Spark SQL windowing

Once you’ve defined a window specification you can compute a function over it, as shown in Example 3-28. Spark’s existing aggregate functions, covered in “Aggregates and groupBy”, can be computed on an aggregation over the window. Window operations are very useful for things like Kalman filtering or many types of relative analysis.

    val pandaRelativeSizeCol = pandas("pandaSize") -
      avg(pandas("pandaSize")).over(windowSpec)

    pandas.select(pandas("name"), pandas("zip"), pandas("pandaSize"), pandas("age"),
      pandaRelativeSizeCol.as("panda_relative_size"))

Sorting

Sorting supports multiple columns in ascending or descending order, with ascending as the default. These sort orders can be intermixed, as shown in Example 3-29. Spark SQL has some extra benefits for sorting as some serialized data can be compared without deserialization.

    pandas.orderBy(pandas("pandaSize").asc, pandas("age").desc)

When limiting results, sorting is often used to only bring back the top or bottom K results. When limiting you specify the number of rows with limit(numRows) to restrict the number of rows in the DataFrame. Limits are also sometimes used for debugging without sorting to bring back a small result. If, instead of limiting the number of rows based on a sort order, you want to sample your data, “Sampling” covers techniques for Spark SQL sampling as well.

Set-like operations

The DataFrame set-like operations allow us to perform many operations that are most commonly thought of as set operations. These operations behave a bit differently than traditional set operations since we don’t have the restriction of unique elements. While you are likely already familiar with the results of set-like operations from regular Spark and Learning Spark, it’s important to review the cost of these operations in Table 3-9.

JSON

Loading and writing JSON is supported directly in Spark SQL, and despite the lack of schema information in JSON, Spark SQL is able to infer a schema for us by sampling the records. Loading JSON data is more expensive than loading many data sources, since Spark needs to read some of the records to determine the schema information. If the schema between records varies widely (or the number of records is very small), you can increase the percentage of records read to determine the schema by setting samplingRatio to a higher value, as in Example 3-33 where we set the sample ratio to 100%.

    val df2 = session.read.format("json")
      .option("samplingRatio", "1.0").load(path)

Since our input may contain some invalid JSON records we may wish to filter out, we can also take in an RDD of strings. This allows us to load the input as a standard text file, filter out our invalid records, and then load the data into JSON. This is done by using the built-in json function on the DataFrameReader, which takes RDDs or paths and is shown in Example 3-34. Methods for converting RDDs of regular objects are covered in “RDDs”.

    val rdd: RDD[String] = input.filter(_.contains("panda"))
    val df = session.read.json(rdd)

JDBC

The JDBC data source represents a natural Spark SQL data source, one that supports many of the same operations. Since different database vendors have slightly different JDBC implementations, you need to add the JAR for your JDBC data sources. Since SQL field types vary as well, Spark uses JdbcDialects with built-in dialects for DB2, Derby, MsSQL, MySQL, Oracle, and Postgres.⁵

While Spark supports many different JDBC sources, it does not ship with the JARs required to talk to all of these databases. If you are submitting your Spark job with spark-submit you can download the required JARs to the host you are launching and include them by specifying --jars or supply the Maven coordinates to --packages. Since the Spark Shell is also launched this way, the same syntax works and you can use it to include the MySQL JDBC JAR in Example 3-35.

spark-submit --jars ./resources/mysql-connector-java-5.1.38.jar $ASSEMBLY_JAR $CLASS

JdbcDialects allow Spark to correctly map the JDBC types to the corresponding Spark SQL types. If there isn’t a JdbcDialect for your database vendor, the default dialect will be used, which will likely work for many of the types. The dialect is automatically chosen based on the JDBC URL used.

As with the other built-in data sources, there exists a convenience wrapper for specifying the properties required to load JDBC data, illustrated in Example 3-36. The convenience wrapper JDBC accepts the URL, table, and a java.util.Properties object for connection properties (such as authentication information). The properties object is merged with the properties that are set on the reader/writer itself. While the properties object is required, an empty properties object can be provided and properties instead specified on the reader/writer.

    session.read.jdbc("jdbc:dialect:serverName;user=user;password=pass",
      "table", new Properties)

    session.read.format("jdbc")
      .option("url", "jdbc:dialect:serverName")
      .option("dbtable", "table").load()

The API for saving a DataFrame is very similar to the API used for loading. The save() function needs no path since the information is already specified, as illustrated in Example 3-37, just as with loading.

    df.write.jdbc("jdbc:dialect:serverName;user=user;password=pass",
      "table", new Properties)

    df.write.format("jdbc")
      .option("url", "jdbc:dialect:serverName")
      .option("user", "user")
      .option("password", "pass")
      .option("dbtable", "table").save()

In addition to reading and writing JDBC data sources, Spark SQL can also run its own JDBC server (covered in “JDBC/ODBC Server”).

Parquet

Apache Parquet files are a common format directly supported in Spark SQL, and they are incredibly space-efficient and popular. Apache Parquet’s popularity comes from a number of features, including the ability to easily split across multiple files, compression, nested types, and many others discussed in the Parquet documentation. Since Parquet is such a popular format, there are some additional options available in Spark for the reading and writing of Parquet files. These options are listed in Table 3-10. Unlike third-party data sources, these options are mostly configured on the SQLContext, although some can be configured on either the SQLContext or DataFrameReader/Writer.

Reading Parquet from an old version of Spark requires some special options, as shown in Example 3-38.

  def loadParquet(path: String): DataFrame = {
    // Configure Spark to read binary data as string,
    // note: must be configured on session.
    session.conf.set("spark.sql.parquet.binaryAsString", "true")

    // Load parquet data using merge schema (configured through option)
    session.read
      .option("mergeSchema", "true")
      .format("parquet")
      .load(path)
  }

Writing parquet with the default options is quite simple, as shown in Example 3-39.

  def writeParquet(df: DataFrame, path: String) = {
    df.write.format("parquet").save(path)
  }

Hive tables

Interacting with Hive tables adds another option beyond the other formats. As covered in “Plain Old SQL Queries and Interacting with Hive Data”, one option for bringing in data from a Hive table is writing a SQL query against it and having the result as a DataFrame. The DataFrame’s reader and writer interfaces can also be used with Hive tables, as with the rest of the data sources, as illustrated in Example 3-40.

  def loadHiveTable(): DataFrame = {
    session.read.table("pandas")
  }

Saving a managed table is a bit different, and is illustrated in Example 3-41.

  def saveManagedTable(df: DataFrame): Unit = {
    df.write.saveAsTable("pandas")
  }

RDDs

Spark SQL DataFrames can easily be converted to RDDs of Row objects, and can also be created from RDDs of Row objects as well as JavaBeans, Scala case classes, and tuples. For RDDs of strings in JSON format, you can use the methods discussed in “JSON”. Datasets of type T can also easily be converted to RDDs of type T, which can provide a useful bridge for DataFrames to RDDs of concrete case classes instead of Row objects. RDDs are a special-case data source, since when going to/from RDDs, the data remains inside of Spark without writing out to or reading from an external system.

When you create a DataFrame from an RDD, Spark SQL needs to add schema information. If you are creating the DataFrame from an RDD of case classes or plain old Java objects (POJOs), Spark SQL is able to use reflection to automatically determine the schema, as shown in Example 3-42. You can also manually specify the schema for your data using the structure discussed in “Basics of Schemas”. This can be especially useful if some of your fields are not nullable. You must specify the schema yourself if Spark SQL is unable to determine the schema through reflection, such as an RDD of Row objects (perhaps from calling .rdd on a DataFrame to use a functional transformation, as shown in Example 3-42).

  def createFromCaseClassRDD(input: RDD[PandaPlace]) = {
    // Create DataFrame explicitly using session and schema inference
    val df1 = session.createDataFrame(input)

    // Create DataFrame using session implicits and schema inference
    val df2 = input.toDF()

    // Create a Row RDD from our RDD of case classes
    val rowRDD = input.map(pm => Row(pm.name,
      pm.pandas.map(pi => Row(pi.id, pi.zip, pi.happy, pi.attributes))))

    val pandasType = ArrayType(StructType(List(
      StructField("id", LongType, true),
      StructField("zip", StringType, true),
      StructField("happy", BooleanType, true),
      StructField("attributes", ArrayType(FloatType), true))))

    // Create DataFrame explicitly with specified schema
    val schema = StructType(List(StructField("name", StringType, true),
      StructField("pandas", pandasType)))

    val df3 = session.createDataFrame(rowRDD, schema)
  }

Converting a DataFrame to an RDD is incredibly simple; however, you get an RDD of Row objects, as shown in Example 3-43. Since a row can contain anything, you need to specify the type (or cast the result) as you fetch the values for each column in the row. With Datasets you can directly get back an RDD templated on the same type, which can make the conversion back to a useful RDD much simpler.

  def toRDD(input: DataFrame): RDD[RawPanda] = {
    val rdd: RDD[Row] = input.rdd
    rdd.map(row => RawPanda(row.getAs[Long](0), row.getAs[String](1),
      row.getAs[String](2), row.getAs[Boolean](3), row.getAs[Array[Double]](4)))
  }

Local collections

Much like with RDDs, you can also create DataFrames from local collections and bring them back as local collections, as illustrated in Example 3-44. The same memory requirements apply; namely, the entire contents of the DataFrame will be in-memory in the driver program. As such, distributing local collections is normally limited to unit tests, or joining small datasets with larger distributed datasets.

  def createFromLocal(input: Seq[PandaPlace]) = {
    session.createDataFrame(input)
  }

Collecting data back as a local collection is more common and often done post aggregations or filtering on the data. For example, with ML pipelines collecting the coefficents or (as discussed in our Goldilocks example in Chapter 6) collecting the quantiles to the driver. Example 3-45 shows how to collect a DataFrame back locally. For larger datasets, saving to an external storage system (such as a database or HDFS) is recommended.

  def collectDF(df: DataFrame) = {
    val result: Array[Row] = df.collect()
    result
  }

Additional formats

As with core Spark, the data formats that ship directly with Spark only begin to scratch the surface of the types of systems with which you can interact. Some vendors publish their own implementations, and many are published on Spark Packages. As of this writing there are over twenty formats listed on the Data Source’s page with the most popular being Avro, Redshift, CSV,⁶ and a unified wrapper around 6+ databases called deep-spark.

Spark packages can be included in your application in a few different ways. During the exploration phase (e.g., using the shell) you can include them by specifying --packages on the command line, as in Example 3-46. The same approach can be used when submitting your application with spark-submit, but this only includes the package at runtime, not at compile time. For including at compile time you can add the Maven coordinates to your builds, or, if building with sbt, the sbt-spark-package plug-in simplifies package dependencies with spDependencies. Otherwise, manually listing them as in Example 3-47 works quite well.

./bin/spark-shell --packages com.databricks:spark-csv_2.11:1.5.0

"com.databricks" % "spark-csv_2.11" % "1.5.0"

Once you’ve included the package with your Spark job you need to specify the format, as you did with the Spark provided ones. The name should be mentioned in the package’s documentation. For spark-csv you would specify a format string of com.databricks.spark.csv. For the built-in CSV format (in Spark 2.0+) you would instead just use csv (or the full name org.apache.spark.sql.csv).

There are a few options if the data format you are looking for isn’t directly supported in either Spark or one of the libraries. Since many formats are available as Hadoop input formats, you can try to load your data as a Hadoop input format and convert the resulting RDD as discussed in “RDDs”. This approach is relatively simple, but means Spark SQL is unable to push down operations to our data store.⁷

For a deeper integration you can implement your data source using the Data Source API. Depending on which operations you wish to support operator push-down for, in your base relation you will need to implement additional traits from the org.apache.spark.sql.sources package. The details of implementing a new Spark SQL data source are beyond the scope of this book, but if you are interested the Scaladoc for org.apache.spark.sql.sources and spark-csv’s CsvRelation can be good ways to get started.