Spark using Python

Pyspark — Cheatsheet

Spark is a powerful open-source distributed computing system designed for big data processing and analytics. Its architecture is built around a few key components that work together to handle large-scale data processing tasks efficiently. Here’s a breakdown of the Spark architecture:

Cluster Manager: Spark can run on various cluster managers like Apache Mesos, Hadoop YARN, or its built-in standalone cluster manager. The cluster manager allocates resources (CPU, memory) across applications.

Driver Program: The driver program is the main entry point for Spark applications. It runs the user’s main function and orchestrates the execution of the tasks in the application. It splits the application into tasks and schedules them on executors.

Executor: Executors are worker nodes in the Spark cluster that are responsible for executing tasks. Each executor runs multiple tasks over its lifetime and holds data in memory or disk storage across them. Executors are launched at the beginning of a Spark application and typically persist for the entire duration of the application.

Get Manpreet Buttar’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

RDD (Resilient Distributed Dataset): RDD is the fundamental data structure in Spark. It represents an immutable, partitioned collection of records that can be operated on in parallel. RDDs are created either by loading data from stable storage (e.g., HDFS, S3) or transforming other RDDs through operations like map, filter, join, etc.

# Parallelzing Collections
from pyspark import SparkContext #Import SparkContext
sc = SparkContext() #initiating a SparkContext object

my_list= [1,2,3,4,5,6,7,8,9,10]
rdd_list = sc.parallelize(my_list,numSlices = 4)
type(rdd_list) # to check type of list
Check number of partitions
rdd_list.getNumPartitions()
rdd_list.glom().collect() # to read the partitioned list
Country_list = ["USA","Canada","India","UK","Russia","England","Pakistan","China"]
rdd_list = sc.parallelize(Country_list, numSlices=2)
rdd_list.getNumPartitions()
rdd_list.glom().collect()
# [['USA', 'Canada', 'India', 'UK'], ['Russia', 'England', 'Pakistan', 'China']]
# external File - make sure file is available on hadoop server
cars_list = sc.textFile('testfolder/AutoInsurance.csv',minPartitions=4)
cars_list.getNumPartitions()

Transformations and Actions: Spark supports two types of operations on RDDs: transformations and actions. Transformations create a new RDD from an existing one (e.g., map, filter, reduceByKey), while actions compute a result based on an RDD and either return it to the driver program or save it to an external storage system (e.g., collect, saveAsTextFile).

### RDD
# Transformations
> map - map is a transformation operation that applies a function to each element of an RDD and returns a new RDD consisting of the results.
A_india_map = A_india.map(lambda x : x.split(' ')) # does same transformation on each object

> flatmap - flatMap is similar to map, but it flattens the results. It applies a function to each element of an RDD and returns a new RDD consisting of the elements of the results.
A_india_flatmap = A_india.flatMap(lambda x : x.split(' ')) # overall give single list instaed of list within list like map

> filter - filter is a transformation operation that applies a predicate function to each element of an RDD and returns a new RDD consisting of the elements that satisfy the predicate. 
A_india_only = A_india_map.filter(lambda x: x[2] == 'India')

> Distinct - distinct is a transformation operation that removes duplicate elements from an RDD and returns a new RDD with only unique elements.
A_india_places_distinct = A_india_places.distinct() #find unique in the list

> Union - union is a transformation operation that combines two RDDs into a single RDD by merging their elements. 
union_places = A_dubai_places_distinct.union(A_india_places_distinct)

> Intersection - intersection is a transformation operation that returns an RDD containing only the elements that are present in both input RDDs. 
intersect_places = A_dubai_places_distinct.intersection(A_india_places_distinct)

# Actions - actions trigger the execution of the RDD transformations that define the computation
> collect - collect is an action operation that retrieves all elements of an RDD and brings them back to the driver program as an array or list in the programming language being used (e.g., Python, Scala). 
A_india.collect()

> take - take is an action operation that retrieves a specified number of elements from an RDD and returns them to the driver program.
A_india.take(5) # like head in pandas

> count - count is an action operation that returns the number of elements in an RDD.
intersect_places.count()

### Pair RDD
# Transformations
student_marks_paired = student_marks_split.map(lambda x: (x[0], (x[1],x[2],x[3])))
student_marks_paired.collect() # to create/view the keyvalue pair

> Keys
student_marks_key = student_marks_paired.keys()
student_marks_paired.collect()  # to view the keys only

> Values
student_marks_values = student_marks_paired.values()
student_marks_paired.collect() # to view values only

> GroupByKey
SM_group = student_marks_1.groupByKey()
SM_group_2 = SM_group.mapValues(sum).collect()

> ReduceByKey
Red_marks = student_marks_1.reduceByKey(lambda x,y : x+y)
Red_marks.collect()
# Note: Less number of shuffle in reduceByKey than groupByKey

> Join - join over the key
list1.join(list2)
list1.leftOuterJoin(list2)
list1.rightOuterJoin(list2)

# Actions
> CountByKey - to give count w.r.t keys
student_marks_key.countByKey()
# defaultdict(int, {'G': 2, 'D': 5, 'F': 2})

student_marks_distinct_cities = student_marks_split.map(lambda x: x[3]).distinct()

result = []

for city in student_marks_distinct_cities.collect():
    city_count = student_marks_split.filter(lambda x : x[3] ==city).count()
    result.append((city,city_count))
    
print(result)

> Lookup
student_marks_paired.lookup("102") # to find values for a defined key

DAG (Directed Acyclic Graph): Spark breaks down a user’s Spark job into smaller tasks and organizes them into a directed acyclic graph (DAG) of stages. Each stage represents a set of tasks that can be executed together, with each task corresponding to a partition of the data.

Schedulers: Spark includes two types of schedulers: a task scheduler and a job scheduler. The task scheduler runs tasks on executors, while the job scheduler manages the execution of stages within a job. The default scheduler is FIFO, but Spark also supports fair scheduling and custom schedulers.

Caching and Persistence: Spark allows users to persist RDDs in memory or disk storage across operations for faster access in subsequent computations. This is particularly useful for iterative algorithms or when RDDs are reused multiple times in a workflow.