Working with Spark SQL DataFrames and Datasets

Table of Contents- Spark SQL - - - DataFrame - Dataset - Project Setup with IDEA - Custom Functions - User Defined Aggregate Functions (Weakly Typed) - User Defined Aggregate Functions (Strongly Typed)

DataFrame Operations

Initializing Spark Session

scala> spark
res0: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@4e8598d9

Exploring available data sources for DataFrame creation

scala> spark.read.
csv   format   jdbc   json   load   option   options   orc   parquet   schema   table   text   textFile

Reading JSON file into DataFrame

scala> val df = spark.read.json("hdfs://192.168.83.101:9000/datas/2.json")
res2: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

Displaying DataFrame contents

scala> df.show()
+---+--------+
|age|    name|
+---+--------+
| 20|zhangsan|
| 20|    lisi|
| 20|  wangwu|
+---+--------+

Creating temporary view (read-only)

scala> df.createTempView("emp")

Querying the temporary view

scala> spark.sql("select * from emp").show
+---+--------+
|age|    name|
+---+--------+
| 20|zhangsan|
| 20|    lisi|
| 20|  wangwu|
+---+--------+

Temporary tables exist within the session scope; upon session termination, they disappear. For application-wide visibility, use global tables with full path access like: global_temp.people

DSL syntax usage

scala> df.printSchema
root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)

Selecting name column data

scala> df.select("name").show()
+--------+
|    name|
+--------+
|zhangsan|
|    lisi|
|  wangwu|
+--------+

Selecting name and age+1 values

scala> df.select($"name",$"age" + 1).show()
+--------+---------+
|    name|(age + 1)|
+--------+---------+
|zhangsan|       21|
|    lisi|       21|
|  wangwu|       21|
+--------+---------+

Filtering records where age is less than 21

scala> df.filter($"age" < 21).show()
+---+--------+
|age|    name|
+---+--------+
| 20|zhangsan|
| 20|    lisi|
| 20|  wangwu|
+---+--------+

Grouping by age and counting entries

scala> df.groupBy("age").count().show()
+---+-----+
|age|count|
+---+-----+
| 21|    2|
| 20|    2|
+---+-----+

Converting RDD to DataFrame

scala> val rdd = sc.makeRDD(List(1,2,3,4))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[40] at makeRDD at <console>:24

scala> val df = rdd.toDF("id")
df: org.apache.spark.sql.DataFrame = [id: int]

scala> df.show()
+---+
| id|
+---+
|  1|
|  2|
|  3|
|  4|
+---+

Transforming a single RDD element to DataFrame

scala> val rdd = sc.makeRDD(List((1,"zhangsan",20),(2,"lisi",30),(3,"wangwu",40)))
rdd: org.apache.spark.rdd.RDD[(Int, String, Int)] = ParallelCollectionRDD[45] at makeRDD at <console>:24

scala> val df = rdd.toDF("id","name","age")
df: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]

scala> df.show()
+---+--------+---+
| id|    name|age|
+---+--------+---+
|  1|zhangsan| 20|
|  2|    lisi| 30|
|  3|  wangwu| 40|
+---+--------+---+

Creating DataFrame via case class

scala> case class People(name:String,age:Int)
defined class People

scala> val rdd = sc.makeRDD(List(("zhangsan",20),("lisi",30),("wangwu",40)))
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[55] at makeRDD at <console>:24

scala> val peopleRDD = rdd.map(t=>{People(t._1,t._2)})
peopleRDD: org.apache.spark.rdd.RDD[People] = MapPartitionsRDD[56] at map at <console>:28

scala> peopleRDD.toDF
res25: org.apache.spark.sql.DataFrame = [name: string, age: int]

Converting DataFrame back to RDD

Direct method call suffices

scala> df.rdd
res27: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[64] at rdd at <console>:29

Dataset Operations

Defining a case class

scala> case class Person(name:String,age:Long)
defined class Person

Creating Dataset

scala> val caseClassDS = Seq(Person("Andy",32)).toDS()
caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

Accessing Dataset (similar to DataFrame)

scala> caseClassDS.show()
+----+---+
|name|age|
+----+---+
|Andy| 32|
+----+---+

Converting RDD to Dataset

scala> case class Person(name:String,age:Long)
defined class Person

scala> val mapRDD = rdd.map(t=>{Person(t._1,t._2)})
mapRDD: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[65] at map at <console>:28

scala> mapRDD.toDS
res30: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

scala> mapRDD.toDS.show()
+--------+---+
|    name|age|
+--------+---+
|zhangsan| 20|
|    lisi| 30|
|  wangwu| 40|
+--------+---+

Dataset to RDD conversion

scala> val ds = mapRDD.toDS
ds: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

scala> ds.rdd
res32: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[73] at rdd at <console>:33

DataFrame to Dataset conversion

scala> rdd.collect
res33: Array[(String, Int)] = Array((zhangsan,20), (lisi,30), (wangwu,40))

scala> val df = rdd.toDF("name","age")
df: org.apache.spark.sql.DataFrame = [name: string, age: int]

scala> df.as[Person]
res34: org.apache.spark.sql.Dataset[Person] = [name: string, age: int]

Dataset to DataFrame conversion

scala> val ds = df.as[Person]
ds: org.apache.spark.sql.Dataset[Person] = [name: string, age: int]

scala> ds.toDF
res35: org.apache.spark.sql.DataFrame = [name: string, age: int]

Project Setup Using IDEA

package com.bigdata.sparksql

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

object SparkSQL01_Demo {
  def main(args: Array[String]): Unit = {
    //TODO Create configuration object
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Demo")
    val sc = new SparkContext(sparkConf)
    //TODO Initialize SparkSQL environment
    val spark = SparkSession.builder().config(sparkConf).config("spark.sql.warehouse.dir", "C:/Users/RenZetong/Desktop/桌面备份/SparkSQLDev/spark-warehouse").getOrCreate()
    //TODO Construct DataFrame
    val frame = spark.read.json("input/2.json")
    frame.show()
    spark.stop()
  }
}

Building DataFrame with SQL Queries

package com.bigdata.sparksql

import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}

object SparkSQL02_SQL {
  def main(args: Array[String]): Unit = {
    //TODO Create configuration object
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Demo")
    val sc = new SparkContext(sparkConf)
    //TODO Initialize SparkSQL environment
    val spark = SparkSession.builder().config(sparkConf).config("spark.sql.warehouse.dir", "C:/Users/RenZetong/Desktop/桌面备份/SparkSQLDev/spark-warehouse").getOrCreate()
    //TODO Construct DataFrame
    val frame = spark.read.json("input/2.json")
    //Convert frame to table
    frame.createTempView("user")
    //Execute SQL query
    spark.sql("select * from user").show()
    spark.stop()
  }
}

Type Conversiosn

package com.bigdata.sparksql

import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}

object SparkSQL03_Transform {
  def main(args: Array[String]): Unit = {
    //TODO Create configuration object
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Demo")
    val sc = new SparkContext(sparkConf)
    //TODO Initialize SparkSQL environment
    val spark = SparkSession.builder().config(sparkConf).config("spark.sql.warehouse.dir", "C:/Users/RenZetong/Desktop/桌面备份/SparkSQLDev/spark-warehouse").getOrCreate()

    //Import implicit conversions
    import spark.implicits._

    //Create RDD
    val rdd = spark.sparkContext.makeRDD(List((1,"张三",22),(2,"李四",20),(3,"王五",30)))
    //Convert to DataFrame
    val df = rdd.toDF("id","name","age")
    //Convert to Dataset
    val ds = df.as[User]
    //Convert back to DataFrame
    val df1 = ds.toDF()
    //Convert to RDD
    val rdd1 = df1.rdd

    rdd1.foreach(
      row=>{
        println(row.getInt(0) + "," + row.getString(1) + "," + row.getInt(2))
      }
    )
  }
}
//TODO Define case class for table structure
case class User(id:Int,name:String,age:Int)

Custom Functions

In Shell, custom functions can be registered through spark.udf

scala> var df = spark.read.json("hdfs://192.168.83.101:9000/datas/2.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

scala> spark.udf.register("addName",(x:String)=>"Name:" + x)
res1: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,StringType,Some(List(StringType)))

scala> df.createTempView
   def createTempView(viewName: String): Unit

scala> df.createTempView("user")
+---+--------+
|age|    name|
+---+--------+
| 20|zhangsan|
| 20|    lisi|
| 21|  wangwu|
| 21|   liliu|
| 18|   liyue|
+---+--------+

scala> spark.sql("select * from user").show()

scala> spark.sql("select addName(name)  from user").show()
+-------------+
|    UDF(name)|
+-------------+
|Name:zhangsan|
|    Name:lisi|
|  Name:wangwu|
|   Name:liliu|
|   Name:liyue|
+-------------+

User Defined Aggregate Functions (Weakly Typed)

Spark provides built-in aggregate functions like count(), countDistinct(), avg(), max(), min(). Users may also define their own.

Weakly typed UDAF implementation involves extending UserDefinedAggregateFunction:

package com.bigdata.sparksql

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, DoubleType, LongType, StructType}

object SparkSQL01_Demo {
  def main(args: Array[String]): Unit = {
    //TODO Create configuration object
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Demo")
    val sc = new SparkContext(sparkConf)
    //TODO Initialize SparkSQL environment
    val spark = SparkSession.builder().config(sparkConf).config("spark.sql.warehouse.dir", "C:/Users/RenZetong/Desktop/桌面备份/SparkSQLDev/spark-warehouse").getOrCreate()
    //TODO Register custom aggregate function
    val udaf = new MyAgeAvgFunction
    spark.udf.register("avgAge", udaf)

    //Use aggregate function
    val frame = spark.read.json("input/2.json")

    frame.createTempView("user")
    spark.sql("select avgAge(age) from user").show()

    spark.stop()
  }
}

//TODO Declare weakly typed UDAF
//1. Extend UserDefinedAggregateFunction
//2. Implement methods
class MyAgeAvgFunction extends UserDefinedAggregateFunction {

  //TODO Input schema definition
  override def inputSchema: StructType = {
    //Input has one field 'age' of type Long
    new StructType().add("age", LongType)
  }

  //TODO Buffer schema definition
  override def bufferSchema: StructType = {
    new StructType().add("sum", LongType).add("count", LongType)
  }

  //TODO Output data type
  override def dataType: DataType = DoubleType

  //TODO Deterministic property
  override def deterministic: Boolean = true

  //TODO Initialize buffer
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0L // Represents sum
    buffer(1) = 0L // Represents count
  }

  //TODO Update buffer based on input
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    buffer(0) = buffer.getLong(0) + input.getLong(0)
    buffer(1) = buffer.getLong(1) + 1
  }

  //TODO Merge buffers from different nodes
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }

  //TODO Final computation result
  override def evaluate(buffer: Row): Any = {
    buffer.getLong(0).toDouble / buffer.getLong(1)
  }
}

User Defined Aggregate Functions (Strongly Typed)

package com.bigdata.sparksql

import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{Dataset, Encoder, Encoders, Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

object SparkSQL04_UDAF_Class {
  def main(args: Array[String]): Unit = {
    //TODO Create configuration object
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Demo")
    val sc = new SparkContext(sparkConf)
    //TODO Initialize SparkSQL environment
    val spark = SparkSession.builder().config(sparkConf).config("spark.sql.warehouse.dir", "C:/Users/RenZetong/Desktop/桌面备份/SparkSQLDev/spark-warehouse").getOrCreate()
    //Import implicit conversions
    import spark.implicits._

    //Create UDAF instance
    val udaf = new MyAgeAvgClassFunction
    //Convert UDAF to column expression
    val vagCol = udaf.toColumn.name("avgAge")

    val frame = spark.read.json("input/2.json")

    val userDS = frame.as[UserBean]

    //Apply function
    userDS.select(vagCol).show()

    spark.stop()
  }
}

case class UserBean(name: String, age: BigInt)

case class AvgBuffer(var sum: BigInt, var count: Int)

//TODO Declare strongly typed UDAF
//1. Extend Aggregator
//2. Implement methods

class MyAgeAvgClassFunction extends Aggregator[UserBean, AvgBuffer, Double] {
  //TODO Initialization
  override def zero: AvgBuffer = {
    AvgBuffer(0, 0)
  }

  /**
   * Aggregate data
   */
  override def reduce(b: AvgBuffer, a: UserBean): AvgBuffer = {
    b.sum = b.sum + a.age
    b.count = b.count + 1
    b
  }

  //TODO Buffer merging
  override def merge(b1: AvgBuffer, b2: AvgBuffer): AvgBuffer = {
    b1.sum = b1.sum + b2.sum
    b1.count = b1.count + b2.count
    b1
  }

  //TODO Final calculation
  override def finish(reduction: AvgBuffer): Double = {
    reduction.sum.toDouble / reduction.count
  }

  //TODO Buffer encoder
  override def bufferEncoder: Encoder[AvgBuffer] = Encoders.product

  //TODO Output encoder
  override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}