Welcome to Apache Spark¶
Architecture¶
A Spark program consists of a driver application and worker programs
- Worker nodes run on different machines in a cluster, or in local threads.
- Data is distributed among workers.
Spark Context¶
The SparkContext contains all of the necessary info on the cluster to run Spark code.
from pyspark import SparkContext, SparkConf
conf = SparkConf().setAppName('spark-app').setMaster('local[*]')
sc = SparkContext.getOrCreate(conf=conf)
sc
Resilient Distributed Dataset¶
A partitioned collection of objects spread accross a cluster, stored in memory or on disk.
3 ways of creating a RDD
- by parallelizing an existing collection
array = range(10)
array
rdd = sc.parallelize(array)
rdd
3 ways of creating a RDD
- from files in a storage system
titanic = sc.textFile('data/titanic.csv')
titanic
titanic.take(3)
3 ways of creating a RDD
- by transforming another RDD
rdd.map(lambda number: number * 2)
rdd.map(lambda number: number * 2).collect()
Working with RDDs¶
Let's create a RDD from a list of numbers, and play with it.
rdd = sc.parallelize(range(16), 4)
rdd.cache()
Remember !
- A RDD is immutable
print(rdd) # prints only info on RDD, no evaluation
- A RDD is evaluated lazily
print(rdd.map(lambda x: x*2)) # specific methods to gather data back to driver
- Only tracks its lineage so it can reconstruct itself
print(rdd.map(lambda num: num + 1).toDebugString()) # check RDD lineage
Spark operations¶
Come in two types : transformations / actions
- Transformations are lazy (not computed immediately)
- Only an action on a RDD will trigger the execution of all subsequent transformations.
Transformations¶
Transformations shape your dataset
Filter¶
Return a new RDD containing only the elements that satisfy a predicate.
Ex : return only even numbers.
rdd.filter(lambda x: x % 2 == 0).collect()
Map¶
Return a new RDD by applying a function to each element of this RDD.
Ex : multiply all numbers by 2.
rdd.map(lambda x: x * 2).collect()
FlatMap¶
Return a new RDD by first applying a function to all elements of this RDD, and then flattening the results.
Ex : return a long matrix of rows [1, 2, 3] of dimension the number of elements in the rdd variable, then flatten it.
rdd.flatMap(lambda num: [1, 2, 3]).take(6)
Distinct¶
Return a new RDD containing the distinct elements in this RDD.
rdd.map(lambda num: 0 if num % 2 == 0 else 1).distinct().collect()
Actions¶
Actions execute the task and associated transformations
Collect / take¶
Return a list that contains all of the elements in this RDD.
Note this method should only be used if the resulting array is expected to be small, as all the data is loaded into the driver’s memory
rdd.take(5)
Count¶
Return the number of elements in this RDD.
rdd.count()
Reduce¶
Reduces the elements of this RDD using the specified commutative and associative binary operator. Currently reduces partitions locally.
Ex : sum all the numbers in the RDD.
rdd.reduce(lambda x,y: x + y)
Key-value transformations¶
Key/value RDDs are commonly used to perform aggregations, and often we will do some initial ETL (extract, transform, and load) to get our data into a key/value format.
Key/value RDDs expose new operations (e.g., counting up reviews for each product, grouping together data with the same key, and grouping together two different RDDs).
ReduceByKey¶
Merge the values for each key using an associative and commutative reduce function.
Ex : Add all numbers associated to each key.
rdd = sc.parallelize([('a', 1), ('b', 0), ('b', 2), ('a', 5)], 4)
rdd.reduceByKey(lambda x,y: x + y).collect()
Join¶
Return an RDD containing all pairs of elements with matching keys in self and other.
Ex : Add all numbers associated to vowels and consonants.
countLetter = sc.parallelize([('a', 1), ('b', 6), ('c', 2), ('a', 5)], 4)
defLetter = sc.parallelize([('a', 'vowel'), ('b', 'consonant'), ('c', 'consonant'), ('d', 'consonant')], 4)
countLetter.join(defLetter).map(lambda x: (x[1][1], x[1][0])).reduceByKey(lambda x,y: x + y).collect()
Wordcount !¶
rdd = sc.textFile('data/lorem.txt')
rdd.flatMap(lambda row: [(r, 1) for r in row.split(' ')]).reduceByKey(lambda x,y: x + y).take(6)
RDD conclusion¶
Resilient Distributed Datasets (RDDs) are a distributed collection of immutable JVM objects that allow you to perform calculations very quickly, and they are the backbone of Apache Spark
sc.stop()
Combine SQL, streaming, and complex analytics.¶
Spark powers a stack of libraries including SQL and DataFrames, MLlib for machine learning, GraphX, and Spark Streaming. You can combine these libraries seamlessly in the same application.
SparkSQL¶
This chapter introduces Spark SQL, Spark’s interface for working with structured and semistructured data.
SparkSession¶
The entry point to programming Spark with the Dataset and DataFrame API.
from pyspark import SparkConf
from pyspark.sql import SparkSession
conf = SparkConf().setAppName('spark-app').setMaster('local[*]')
spark = SparkSession.builder.config(conf=conf).getOrCreate()
spark
Dataframes¶
Under the hood, a Dataframe is an RDD composed of Row objects with additional schema information of the types in each col‐ umn. Row objects are just wrappers around arrays of basic types.
titanic = spark.read.option('header', 'true').option('inferSchema', 'true').csv('data/titanic.csv')
titanic.createOrReplaceTempView('titanic')
titanic.show(8)
Two ways of interacting¶
- Domain-specific language for structured data manipulation
titanic.filter(titanic.Sex == 'male').select(['Name', 'Sex', 'Survived']).show(3)
sqlfunction on SparkSession to run SQL queries programmatically on temporary tables
spark.sql('SELECT Name, Sex, Survived FROM titanic WHERE Sex = "male"').show(3)
Unified data source interaction¶
Spark provides with a unique interface for reading/saving data, which is then implemented for multiple data storage formats : json, parquet, jdbc, orc, libsvm, csv, text.
ransomware = spark.read.json('data/ransomware.json')
ransomware.printSchema()
Catalyst optimization¶
Catalyst is an extensible query optimizer used internally by SparkSQL for planning and defining the execution of SparkSQL queries.
titanic[titanic['Sex'] == 'male'].select(['Name', 'Sex']).explain()
Machine Learning¶
MLlib is Spark’s machine learning (ML) library. It has an RDD-based API in maintenance mode and a Dataframe-based API.
- Dataframe API = Spark Datasources, SQL/DataFrame queries, Tungsten and Catalyst optimizations, uniform APIs across languages.
- ML Pipelines are set of high-level APIs on top of DataFrames that help users create and tune practical machine learning pipelines
Transformers¶
A Transformer implements a method transform(), which converts one DataFrame into another
from pyspark.ml.feature import StringIndexer
indexer = StringIndexer(inputCol="Sex", outputCol="SexIndex")
titanic_indexed = indexer.fit(titanic).transform(titanic)
titanic_indexed.show(8)
Estimators¶
An Estimator implements a method fit(), which accepts a DataFrame and produces a Model, which is a Transformer.
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.feature import VectorAssembler
assembler = VectorAssembler(inputCols=["SexIndex", "Fare"], outputCol="features")
titanic_train = assembler.transform(titanic_indexed)
rf = RandomForestClassifier(labelCol="Survived", featuresCol="features", numTrees=10)
model = rf.fit(titanic_train)
model.transform(titanic_train).select(["Survived", "prediction", "probability"]).show(8)
Pipelines¶
MLlib standardizes APIs for machine learning algorithms to make it easier to combine multiple algorithms into a single pipeline, or workflow.
from pyspark.ml import Pipeline
pipeline = Pipeline(stages=[indexer, assembler, rf])
model = pipeline.fit(titanic)
model.transform(titanic).select(["Survived", "prediction", "probability"]).show(8)
Spark Streaming¶
Spark Streaming is an extension of the core Spark API that enables scalable, high-throughput, fault-tolerant stream processing of live data streams.
# Prepare a netcat client before launching launchSparkStreaming
import nclib
#nc = nclib.Netcat(listen=('localhost', 9999), verbose=True)
#for i in range(1000):
#nc.send_line(b'hello world')
#nc.close()
GraphX¶
To support graph computation, GraphX extends the Spark RDD by introducing a new Graph abstraction: a directed multigraph with properties attached to each vertex and edge.
NB : No active development of Python bindings on GraphX...take a look on GraphFrames for graph computation on Dataframes, which is the unofficial GraphX Dataframe-based API.
Going further¶
Spark packages¶
Unified engine¶
Spark's main contribution is to enable previously disparate cluster workloads to be composed. In the following example, we build a logistic model on the titanic dataset, save it on disk and push it to spark streaming for realtime inference.
spark.stop()
Conclusion¶
