17 SparkSQL

Spark
SQL and DataFrames

Chapter 17
201509
Course Chapters
1 IntroducKon Course IntroducKon

2 IntroducKon to Hadoop and the Hadoop Ecosystem
IntroducKon to Hadoop
3 Hadoop Architecture and HDFS
4 ImporKng RelaKonal Data with Apache Sqoop
5 IntroducKon to Impala and Hive
ImporKng and Modeling Structured
6 Modeling and Managing Data with Impala and Hive
Data
7 Data Formats
8 Data ParKKoning
9 Capturing Data with Apache Flume IngesKng Streaming Data
10 Spark Basics
11 Working with RDDs in Spark
12 AggregaKng Data with Pair RDDs
13 WriKng and Deploying Spark ApplicaKons Distributed Data Processing with
14 Parallel Processing in Spark Spark
15 Spark RDD Persistence
16 Common PaHerns in Spark Data Processing
17 Spark SQL and DataFrames
18 Conclusion Course Conclusion
Copyright 2010-2015 Cloudera. All rights reserved. Not to be reproduced or shared without prior wriHen consent from Cloudera. 17-2
DataFrames and SparkSQL
In this chapter you will learn

What Spark SQL is
What features the DataFrame API provides
How to create a SQLContext
How to load exisIng data into a DataFrame
How to query data in a DataFrame
How to convert from DataFrames to Pair RDDs
Chapter Topics
Distributed Data Processing with

Spark SQL and DataFrames
Spark
Spark SQL and the SQL Context

CreaKng DataFrames
Transforming and Querying DataFrames
Saving DataFrames
DataFrames and RDDs
Comparing Spark SQL, Impala and Hive-on-Spark
Conclusion
Homework: Use Spark SQL for ETL
What is Spark SQL?
What is Spark SQL?

Spark module for structured data processing
Replaces Shark (a prior Spark module, now deprecated)
Built on top of core Spark
What does Spark SQL provide?

The DataFrame API a library for working with data as tables
Denes DataFrames containing Rows and Columns
DataFrames are the focus of this chapter!
Catalyst OpKmizer an extensible opKmizaKon framework
A SQL Engine and command line interface
SQL Context
The main Spark SQL entry point is a SQL Context object

Requires a SparkContext
The SQL Context in Spark SQL is similar to Spark Context in core Spark
There are two implementaIons
SQLContext
basic implementaKon
HiveContext
Reads and writes Hive/HCatalog tables directly
Supports full HiveQL language
Requires the Spark applicaKon be linked with Hive libraries
Recommended starKng with Spark 1.5
CreaKng a SQL Context
SQLContext is created based on the SparkContext
from pyspark.sql import SQLContext

Python
sqlCtx = SQLContext(sc)
import org.apache.spark.sql.SQLContext
Scala val sqlCtx = new SQLContext(sc)
import sqlCtx._
Chapter Topics

Spark

CreaIng DataFrames
Saving DataFrames
DataFrames and RDDs
Conclusion
DataFrames
DataFrames are the main abstracIon in Spark SQL

Analogous to RDDs in core Spark
A distributed collecKon of data organized into named columns
Built on a base RDD containing Row objects
CreaKng DataFrames
DataFrames can be created

From an exisKng structured data source (Parquet le, JSON le, etc.)
From an exisKng RDD
By performing an operaKon or query on another DataFrame
By programmaKcally dening a schema
Example: CreaKng a DataFrame from a JSON File
from pyspark.sql import SQLContext

Python sqlCtx = SQLContext(sc)
peopleDF = sqlCtx.jsonFile("people.json")
val sqlCtx = new SQLContext(sc)

Scala import sqlCtx._
val peopleDF = sqlCtx.jsonFile("people.json")
File: people.json
age name pcode
{"name":"Alice", "pcode":"94304"} null Alice 94304
{"name":"Brayden", "age":30, "pcode":"94304"} 30 Brayden 94304
{"name":"Carla", "age":19, "pcode":"10036"}
{"name":"Diana", "age":46} 19 Carla 10036
{"name":"tienne", "pcode":"94104"}
46 Diana null
null tienne 94104
CreaKng a DataFrame from a Data Source
Methods on the SQLContext object

Convenience funcIons
jsonFile(filename)
parquetFile(filename)
Generic base funcIon: load
load(filename,source) load filename of type source
(default Parquet)
load(source,options) load from a source of type source
using opKons
Convenience funcKons are implemented by calling load
jsonFile("people.json") = load("people.json",
"json")
Data Sources
Spark SQL 1.3 includes three data source types

json
parquet
jdbc
You can also use third party data source libraries, such as
Avro
HBase
CSV
MySQL
and more being added all the Kme
Generic Load FuncKon Example: JDBC
Example: Loading from a MySQL database
val accountsDF = sqlCtx.load("jdbc",

Map("url"-> "jdbc:mysql://dbhost/dbname?user=&password=",
"dbtable" -> "accounts"))
accountsDF = sqlCtx.load(source="jdbc", \
url="jdbc:mysql://dbhost/dbname?user=&password=", \
dbtable="accounts)
Warning: Avoid direct access to databases in producKon environments, which

may overload the DB or be interpreted as service aHacks
Use Sqoop to import instead
Generic Load FuncKon Example: Third-party or Custom Sources
You can also use custom or third party data sources

Example: Read from an Avro le using the avro source in the Databricks
Spark Avro package
$ spark-shell --packages com.databricks:spark-avro_2.10:1.0.0

>
> val myDF =
sqlCtx.load("myfile.avro","com.databricks.spark.avro")
$ pyspark --packages com.databricks:spark-avro_2.10:1.0.0

>
> myDF = sqlCtx.load("myfile.avro","com.databricks.spark.avro")
Chapter Topics

Spark

CreaKng DataFrames
Saving DataFrames
DataFrames and RDDs
Conclusion
DataFrame Basic OperaKons (1)
Basic OperaIons deal with DataFrame metadata (rather than its data), e.g.
schema returns a Schema object describing the data
printSchema displays the schema as a visual tree
cache / persist persists the DataFrame to disk or memory
columns returns an array containing the names of the columns
dtypes returns an array of (column-name,type) pairs
explain prints debug informaKon about the DataFrame to the
console
DataFrame Basic OperaKons (2)
Example: Displaying column data types using dtypes
> peopleDF = sqlCtx.jsonFile("people.json")

> for item in peopleDF.dtypes(): print item
('age', 'bigint')
('name', 'string')
('pcode', 'string)
> val peopleDF = sqlCtx.jsonFile("people.json")

> people.dtypes.foreach(println)
(age,LongType)
(name,StringType)
(pcode,StringType)
Working with Data in a DataFrame
Queries create a new DataFrame

DataFrames are immutable
Queries are analogous to RDD transformaKons
AcIons return data to the Driver
AcKons trigger lazy execuKon of queries
DataFrame AcKons
Some DataFrame acIons

collect return all rows as an array of Row objects
take(n) return the rst n rows as an array of Row objects
count return the number of rows
show(n) display the rst n rows (default=20)
> peopleDF.count() > peopleDF.count()

5L res7: Long = 5
> peopleDF.show(3) > peopleDF.show(3)

age name pcode age name pcode
null Alice 94304 null Alice 94304
30 Brayden 94304 30 Brayden 94304
19 Carla 10036 19 Carla 10036
DataFrame Queries (1)
DataFrame query methods return new DataFrames

Queries can be chained like transformaKons
Some query methods
distinct returns a new DataFrame with disKnct elements of this DF
join joins this DataFrame with a second DataFrame
several variants for inside, outside, lem, right, etc.
limit a new DF with the rst n rows of this DataFrame
select a new DataFrame with data from one or more columns of
the base DataFrame
filter a new DataFrame with rows meeKng a specied condiKon

DataFrame Queries (2)
Example: A basic query with limit

age name pcode

null Alice 94304
> peopleDF.limit(3).show
30 Brayden 94304
19 Carla 10036
> peopleDF.limit(3).show() 46 Diana null
null tienne 94104
age name pcode

Output null Alice 94304 age name pcode
of show 30 Brayden 94304
19 Carla 10036 null Alice 94304
30 Brayden 94304
19 Carla 10036
DataFrame Query Strings (1)
Some query operaIons take strings containing

age
simple query expressions
null
Such as select and where
30
Example: select 19
o p l e DF. e")
46
age name pcode pe t ( " a g
selec
null
null Alice 94304
30 Brayden 94304
name age
19 Carla 10036
people
Alice null
46 Diana null DF. Brayden 30
select
null tienne 94104 ("name
" ,"age" Carla 19
)
Diana 46
tienne null
DataFrame Query Strings (2)
Example: where

peopleDF.
where("age > 21")

null Alice 94304 30 Brayden 94304
30 Brayden 94304 46 Diana null
19 Carla 10036
46 Diana null
null tienne 94104
Querying DataFrames using Columns (1)
Some DF queries take one or more columns or column expressions

Required for more sophisKcated operaKons
Some examples
select
sort
join
where
In Python, reference columns by name using dot nota1on

ageDF = peopleDF.select(peopleDF.age) age name pcode
null Alice 94304
30 Brayden 94304
In Scala, columns can be referenced in two ways 19 Carla 10036

46 Diana null
val ageDF = peopleDF.select($"age") null tienne 94104
OR
age
val ageDF = peopleDF.select(peopleDF("age"))
null
30
19
46
null
Column references can also be column expressions
peopleDF.select(peopleDF.name,peopleDF.age+10)
peopleDF.select(peopleDF("name"),peopleDF("age")+10)
age name pcode name age+10

null Alice 94304 Alice null
30 Brayden 94304 Brayden 40
19 Carla 10036 Carla 29
46 Diana null Diana 56
null tienne 94104 tienne null
Example: SorIng in by columns (descending)

.asc and .desc
are column expression
peopleDF.sort(peopleDF.age.desc())
methods used with
sort
peopleDF.sort(peopleDF("age").desc)

null Alice 94304 46 Diana null
30 Brayden 94304 30 Brayden 94304
19 Carla 10036 19 Carla 10036
46 Diana null null Alice 94304
null tienne 94104 null tienne 94104
SQL Queries
Spark SQL also supports the ability to perform SQL queries

First, register the DataFrame as a table with the SQL Context
peopleDF.registerTempTable("people")
sqlCtx.sql("""SELECT * FROM people WHERE name LIKE "A%" """)
peopleDF.registerTempTable("people")
sqlCtx.sql("""SELECT * FROM people WHERE name LIKE "A%" """)

null Alice 94304 null Alice 94304
30 Brayden 94304
19 Carla 10036
46 Diana null
null tienne 94104
Chapter Topics

Spark

CreaKng DataFrames
Saving DataFrames
DataFrames and RDDs
Conclusion
Saving DataFrames
Data in DataFrames can be saved to a data source

Built in support for JDBC and Parquet File
createJDBCTable create a new table in a database
insertInto save to an exisKng table in a database
saveAsParquetFile save as a Parquet le (including schema)
saveAsTable save as a Hive table (HiveContext only)
Can also use third party and custom data sources
save generic base funcKon
Chapter Topics

Spark

CreaKng DataFrames
Saving DataFrames
DataFrames and RDDs
Conclusion
DataFrames and RDDs (1)
DataFrames are built on RDDs

Base RDDs contain Row objects
Use rdd to get the underlying RDD
peopleRDD = peopleDF.rdd
peopleDF peopleRDD
age name pcode Row[null,Alice,94304]
null Alice 94304 Row[30,Brayden,94304]
30 Brayden 94304 Row[19,Carla,10036]
19 Carla 10036 Row[46,Diana,null]
46 Diana null Row[null,tienne,94104]
null tienne 94104
DataFrames and RDDs (2)
Row RDDs have all the standard Spark acIons and transformaIons
AcKons collect, take, count, etc.
TransformaKons map, flatMap, filter, etc.
Row RDDs can be transformed into PairRDDs to use map-reduce methods
Working with Row Objects
The syntax for extracIng data from Rows depends on language

Python
Column names are object aHributes
row.age return age column value from row
Scala
Use Array-like syntax
row(0) returns element in the rst column
row(1) return element in the second column
etc.
Use type-specic get methods to return typed values
row.getString(n) returns nth column as a String
row.getInt(n) returns nth column as an Integer
etc.
Example: ExtracKng Data from Rows
Row[null,Alice,94304]
Extract data from Rows
Row[30,Brayden,94304]
Row[19,Carla,10036]
peopleRDD = peopleDF.rdd Row[46,Diana,null]
peopleByPCode = peopleRDD \
Row[null,tienne,94104]
.map(lambda row(row.pcode,row.name)) \
.groupByKey()
(94304,Alice)
(94304,Brayden)
(10036,Carla)
val peopleRDD = peopleDF.rdd (null,Diana)

peopleByPCode = peopleRDD. (94104,tienne)
map(row => (row(2),row(1))).
groupByKey()) (null,[Diana])
(94304,[Alice,Brayden])
(10036,[Carla])
(94104,[tienne])
ConverKng RDDs to DataFrames
You can also create a DF from an RDD

sqlCtx.createDataFrame(rdd)
Chapter Topics

Spark

CreaKng DataFrames
Saving DataFrames
DataFrames and RDDs
Conclusion
Comparing Impala to Spark SQL
Spark SQL is built on Spark, a general purpose processing engine

Provides convenient SQL-like access to structured data in a Spark
applicaKon
Impala is a specialized SQL engine
Much beHer performance for querying
Much more mature than Spark SQL
Robust security via Sentry
Impala is beber for
InteracKve queries
Data analysis
Use Spark SQL for
ETL
Access to structured data required by a Spark applicaKon
Comparing Spark SQL with Hive on Spark
Spark SQL
Provides the DataFrame API to allow structured data
processing in a Spark applica-on
Programmers can mix SQL with procedural processing
Hive-on-Spark
Hive provides a SQL abstracKon layer over MapReduce or
Spark
Allows non-programmers to analyze data using familiar
SQL
Hive-on-Spark replaces MapReduce as the engine
underlying Hive
Does not aect the user experience of Hive
Except many Kmes faster queries!
Chapter Topics

Spark

CreaKng DataFrames
Saving DataFrames
DataFrames and RDDs
Conclusion
EssenKal Points
Spark SQL is a Spark API for handling structured and semi-structured data
Entry point is a SQLContext
DataFrames are the key unit of data
DataFrames are based on an underlying RDD of Row objects
DataFrames query methods return new DataFrames; similar to RDD
transformaIons
The full Spark API can be used with Spark SQL Data by accessing the
underlying RDD
Spark SQL is not a replacement for a database, or a specialized SQL engine
like Impala
Spark SQL is most useful for ETL or incorporaKng structured data into
other applicaKons
Chapter Topics

Spark

CreaKng DataFrames
Saving DataFrames
DataFrames and RDDs
Conclusion
In this homework assignment you will

Import the data from MySQL
Use Spark to normalize the data
Save the data to Parquet format
Query the data with Impala or Hive
Please refer to the Homework descripIon

17 SparkSQL

Uploaded by

Copyright:

Available Formats

17 SparkSQL

Uploaded by

Document Information

Original Title

Copyright

Available Formats

Share this document

Share or Embed Document

Sharing Options

Did you find this document useful?

Is this content inappropriate?

Copyright:

Available Formats

17 SparkSQL

Uploaded by

Copyright:

Available Formats

Spark

SQL and DataFrames

1 IntroducKon Course IntroducKon

18 Conclusion Course Conclusion

In this chapter you will learn

Distributed Data Processing with

Spark SQL and the SQL Context

What is Spark SQL?

What does Spark SQL provide?

The main Spark SQL entry point is a SQL Context object

SQLContext is created based on the SparkContext

from pyspark.sql import SQLContext

Distributed Data Processing with

Spark SQL and the SQL Context

DataFrames are the main abstracIon in Spark SQL

DataFrames can be created

from pyspark.sql import SQLContext

val sqlCtx = new SQLContext(sc)

Methods on the SQLContext object

Spark SQL 1.3 includes three data source types

Example: Loading from a MySQL database

val accountsDF = sqlCtx.load("jdbc",

Warning: Avoid direct access to databases in producKon environments, which

You can also use custom or third party data sources

$ spark-shell --packages com.databricks:spark-avro_2.10:1.0.0

$ pyspark --packages com.databricks:spark-avro_2.10:1.0.0

Distributed Data Processing with

Spark SQL and the SQL Context

Example: Displaying column data types using dtypes

> peopleDF = sqlCtx.jsonFile("people.json")

> val peopleDF = sqlCtx.jsonFile("people.json")

Queries create a new DataFrame

Some DataFrame acIons

> peopleDF.count() > peopleDF.count()

> peopleDF.show(3) > peopleDF.show(3)

DataFrame query methods return new DataFrames

Example: A basic query with limit

age name pcode

Some query operaIons take strings containing

age name pcode age name pcode

Some DF queries take one or more columns or column expressions

In Python, reference columns by name using dot nota1on

In Scala, columns can be referenced in two ways 19 Carla 10036

Column references can also be column expressions

age name pcode name age+10

Example: SorIng in by columns (descending)

age name pcode age name pcode

Spark SQL also supports the ability to perform SQL queries

age name pcode age name pcode

Distributed Data Processing with

Spark SQL and the SQL Context

Data in DataFrames can be saved to a data source

Distributed Data Processing with

Spark SQL and the SQL Context

DataFrames are built on RDDs

The syntax for extracIng data from Rows depends on language

val peopleRDD = peopleDF.rdd (null,Diana)

You can also create a DF from an RDD

Distributed Data Processing with