Apache Spark

Lorenzo Di Gaetano

THE CONTRACTOR IS ACTING UNDER A FRAMEWORK CONTRACT CONCLUDED WITH THE COMMISSION

Eurostat
What is Apache Spark?
• A general purpose framework for big data
processing

• It interfaces with many distributed file systems,

such as Hdfs (Hadoop Distributed File System),
Amazon S3, Apache Cassandra and many others

• 100 times faster than Hadoop for in-memory

computation
2
Eurostat
Multilanguage API
• You can write applications in various languages
• Java
• Python
• Scala
• R

• In the context of this course we will consider

Python

3
Eurostat
Built-in Libraries

4
Eurostat
Third party libraries

• Many third party libraries are available

• http://spark-packages.org/

• We used spark-csv in our examples

• We will see later how to use an external jar on
our application

5
Eurostat
Getting started with Spark!

• Spark can run on Hadoop, on the cloud and can

be also be installed as a standalone application
on your PC
• Spark comes pre-installed in all the major
Hadoop distributions
• We will consider the stand-alone installation in
this course

6
Eurostat
Prerequisites

• Python (for using pySpark) and Java must be

installed.

• All environment variables must be correctly set.

7
Eurostat
Local installation on Windows (1/2)
• Download Spark at
https://spark.apache.com/downloads.html
• Download “Pre-built for Hadoop 2.6 and later”
• Unzip it
• Set the SPARK_HOME environment variable to
point where you unzipped Spark
• Add %SPARK_HOME%/bin to your PATH

8
Eurostat
Local installation on Windows (2/2)
• Download http://public-repo-
1.hortonworks.com/hdp-win-alpha/winutils.exe
• Move it in %SPARK_HOME%/bin
• Create an enviroment variable:
HADOOP_HOME=%SPARK_HOME%
• Now test it: open a terminal and launch pySpark

9
Eurostat
PySpark up and running!

10
Eurostat
Running Spark
• Once you correctly installed spark you can use it
in two ways.

• spark-submit: it’s the CLI command you can use to

launch python spark applications

• pyspark: used to launch an interactive python shell.

11
Eurostat
Running Spark

• Using spark-submit you have to manually create

a SparkContext object in your code.

• Using the pyspark interactive shell a

SparkContext is automatically available in a
variable called sc.

12
Eurostat
What is a Spark Application?

• A Spark application is made of a Driver Program

• The Driver Program runs the the main function

which executes parallel operations on the cluster

13
Eurostat
RDD

• Spark works on RDD – Resilient Distributed

Filesystem

• A RDD is a collection of elements partitioned in

every cluster node. Spark operates in parallel on
them

14
Eurostat
RDD

• Every RDD is created from a file on Hadoop

filesystem

• They can be made persistent in memory

15
Eurostat
How to create a RDD
• There are two ways:

• Parallelizing a pre-existent collection on the driver

program

• Reading an external dataset on the filesystem

16
Eurostat
RDD from collections

• You can create a RDD from a collection using

parallelize() method on SparkContext

data = [1,2,3,4,5]
rdd = sc.parallelize(data)

• Then you can operate in parallel on rdd

17
Eurostat
RDD from external datasets

• You can create RDD from various kind of external

datasets like local filesystem, HDFS, Cassandra
etc…
• For example we can read a text file, obtaining a
collection of lines.

rdd = sc.textFile("textfile.txt")

18
Eurostat
Operations on RDD
• There are two types of operations:

• Transformations: Take an existing dataset and return

another dataset

• Actions: Take an existing dataset and return a value

19
Eurostat
Transformations

• For example, map is a transformation that takes

all elements of the dataset, pass them to a
function and returns another RDD with the
results

resultRDD = originalRDD.map(myFunction)

20
Eurostat
Actions

• For example, reduce is an action. It aggregates

all elements of the RDD using a function and
returns the result to the driver program

result = rdd.reduce(function)

21
Eurostat
Passing functions to Spark
• There are three ways

• lambda expressions: used to pass simple expressions

which return a value.

• Local function definition: for more complex code

• Functions inside a module

22
Eurostat
Example: word counting

def countWords(s):
words = s.split(" ")
return len(words)

sc = SparkContext(...)
sc.textFile("textFile.txt").map(countWords)

23
Eurostat
Shared Variables
• We always must keep in mind that when we pass
a function to a spark operation, this function is
executed on separate cluster nodes. Every node
receives a COPY of the variable inside the
function

• Every change to the local value of the variable is

not propagated to the driver program

24
Eurostat
Shared Variables
• To solve the problem spark offers two types of
shared variables:

• Broadcast variables

• Accumulators

25
Eurostat
Broadcast variables
• Instead of creating a copy of the variable for
each machine, broadcast variables allows the
programmer to keep a cached read-only variable
in every machine

• We can create a broadcast variable with the

SparkContext.broadcast(var) method, which
returns the reference of the broadcast variable

26
Eurostat
Accumulators
• Accumulators are used to keep some kind of
shared «counter» across the machines. It’s a
special variable which can be «added»

• We can create an accumulator with

SparkContext.accumulator(var) method

• Once the accumulator is created we can add

values to it with the add() method
27
Eurostat
Making RDD persistent
• Spark can persist (or cache) a dataset in memory
during operations
• If you persist an RDD, every node stores the
partitions it elaborates in memory and reuses
them in other actions which will run faster
• You can persist an RDD using the persist() or
cache() methods

28
Eurostat
Printing elements
• When working on a single machine we can simply
use rdd.foreach(println)
• But when we work in cluster mode, the println
will be executed on the executor stdout, so we
will not see anything on the driver node!
• Instead, we can rdd.take(n).foreach(println) to
print the first n elements of the collection, in
order to be sure we will not run out of memory.

29
Eurostat
Removing data

• Spark perfoms a sort of garbage collection and

automatically deletes old data partitions

• We can manually remove an RDD using the

unpersist() method

30
Eurostat
SparkSQL and DataFrames

• SparkSQL is the spark module for structured data

processing

• DataFrame API is one of the ways to interact with

SparkSQL

31
Eurostat
DataFrames
• A DataFrame is a collection of data organized into
columns

• Similar to tables in relational databases

• Can be created from various sources: structured

data files, Hive Tables, external db, csv etc…

32
Eurostat
Creating a DataFrame
• Given a SparkContext (sc), the first thing to do is
to create a SQLContext

sqlContext = SQLContext(sc)

• Then read the data, for example in JSON format:

df = sqlContext.read.json(‘file.json’)

33
Eurostat
Creating a DataFrame from csv file

• We now see how we can load a csv file into a

DataFrame using an external library called Spark
csv: https://github.com/databricks/spark-csv

• We need to download two files: spark-csv_2.11-

1.4.0.jar and commons-csv-1.2.jar

34
Eurostat
Creating a DataFrame from csv file

• When launching our code with spark-submit we

have to add external jars to the classpath,
assuming we put the jars into lib directory of our
project:

spark-submit --jars lib/spark-csv_2.11-1.4.0.jar,lib/commons-csv-1.2.jar myCode.py

35
Eurostat
Creating a DataFrame from csv file
• Then we read our file into a DataFrame

df = sqlContext.read.format('com.databricks.spark.csv').options(header='true',
nullValue='NULL').load(myCsvFile, inferschema = ‘true’)

• Note the inferschema = true option. With this

option activated, Spark tries to guess the format
of every field
• We can specify the schema manually creating a
StructType array
36
Eurostat
Creating a DataFrame from csv file
• Create the schema structure
customSchema = StructType([ \
StructField("field1", IntegerType(), True),
StructField("field2", FloatType(), True), \
StructField("field3", TimestampType(), True), \
StructField("field4", StringType(), True) \
])

• And then pass it to our load function as an option

df = sqlContext.read.format('com.databricks.spark.csv').options(header='true',
nullValue='NULL').load(myFile, schema = customSchema)

37
Eurostat
Example operations on DataFrames
• To show the content of the DataFrame
• df.show()
• To print the Schema of the DataFrame
• df.printSchema()
• To select a column
• df.select(‘columnName’).show()
• To filter by some parameter
• df.filter(df[‘columnName’] > N).show()

38
Eurostat
A complete example: group and avg
• We have a table like this

+----------+------+-------+--------------+------------------+-----------+-------+
|codice_pdv|prdkey| week|vendite_valore|vendite_confezioni|flag_sconto| sconto|
+----------+------+-------+--------------+------------------+-----------+-------+
| 4567|730716|2015009| 190.8500061| 196.0| 0.0|0.98991|
| 4567|730716|2013048| 174.6000061| 153.0| null| null|
| 4567|730716|2015008| 160.6000061| 165.0| 0.0|0.98951|
| 4567|730716|2015006| 171.92999268| 176.0| 0.0|0.99329|
| 4567|730716|2015005| 209.47999573| 213.0| 0.0|1.00447|
+----------+------+-------+--------------+------------------+-----------+-------+

• We want to group by prdkey and calculate the

average vendite_valore for every group

39
Eurostat
Preparing the schema
sc = SparkContext("local", "PrezzoMedio")

sqlContext = SQLContext(sc)

customSchema = StructType([ \
StructField("codice_pdv", IntegerType(), True), \
StructField("prdkey", IntegerType(), True), \
StructField("week", IntegerType(), True), \
StructField("vendite_valore", FloatType(), True), \
StructField("vendite_confezioni", FloatType(), True), \
StructField("data_ins", TimestampType(), True), \
StructField("num_riga", FloatType(), True), \
StructField("flag_sconto", FloatType(), True), \
StructField("sconto", FloatType(), True), \
StructField("idricefile", FloatType(), True), \
StructField("iddettzipfile", FloatType(), True), \
StructField("uteins", StringType(), True), \
StructField("datamod", TimestampType(), True), \
StructField("utemod", StringType(), True) \
])

40
Eurostat
Read data and do the job
myCsvFile = "C:\TransFatt1000.csv"

df = sqlContext.read.format('com.databricks.spark.csv').options(header='true',
nullValue='NULL').load(myCsvFile, schema = customSchema)

t0 = time()

averageForProductKey = df.groupBy("prdkey").avg("vendite_valore").collect()

tt = time() - t0

print(averageForProductKey)

print ("Query performed in {:03.2f} seconds.".format(tt))

41
Eurostat
With parquet table on HDFS
myParquetTable = "hdfs://bigdata-mnn.hcl.istat.it:8020///user/hive/warehouse/scanner_data.db/trans_fatt_p«

df = sqlContext.read.load(myParquetTable, schema = customSchema)

t0 = time()

averageForProductKey = df.groupBy("prdkey").avg("vendite_valore").collect()

tt = time() - t0

print(averageForProductKey)

print ("Query performed in {:03.2f} seconds.".format(tt))

42
Eurostat
References

• http://spark.apache.org/

• http://spark.apache.org/sql/

• http://spark.apache.org/docs/latest/sql-
programming-guide.html

43
Eurostat