Apache Spark Core APIs— Part 1

Basics

RDD — Resilient distributed dataset, it is the basic unit that holds the data in spark.
Assume you have a 4 node cluster and a file which is divided into 4 blocks, so each node holdes 1 cluster, now when each of these blocks are loaded to memory (RAM) of the worker node it is called partition, so for our use case we have 4 partitions and all these partitions together is called RDD.
So RDD is a distributed collection in memory.

There are two kinds of operations in Apache Spark:
1. Transformation
2. Action

Let’s understand with an example:
Step 1 →Assume the above case a file with 4 data blocks when loaded to memory becomes RDD, say RDD1
Step 2 → now say there are bunch of transformations are happening on this RDD1 (map, filter etc.) and it becomes RDD2
Step 3 → Finally we need to get the results so we run RDD2.collect()
Step 1 & 2 are Transformations
Step 3 is an action
- All transformations are lazy while actions are now, so transformations become action plan and get executed when actions are called
- For every transformation a new RDD is created, and one can’t make change to an existing RDD as RDDs are immutable
- If an RDD (say RDD2) gets failed, system applies the same transformation on the previous RDD (RDD1) and gets the RDD2 back, hence the “R” in RDD means Resilient as it is reslient to failures

Q. What happens if the initial RDD (like RDD1) fails? How does Spark recover it?
If rdd1 is from a file → Spark reads the file again.
If rdd1 is only from memory → Spark cannot recover it if lost.

Now above mentioned steps are called a DAG (directed cyclic graph)

Why are transformations lazy? — Lazy helps to optmise the execution of the DAG, let’s understand the pseudo code
step1 → Load the files (of 1 billion records)
step2 → Perform a map operation
step 3 → Perform a filter operation
step 4 → collect the data or print the data

Now if transformations were not lazy, it would have loaded a file of 1 billion records in the memory, performed map operation on 1 billion rows and then filtered, which would have given say 10 records
But internally what spark does it is filters the records and then perform map operations, so 1 billion records are only processed once and then only filtered records are mapped

Mental note →Write another blog on lazy operations!!

Spark has two level of APIs:
1. Layer 1 of Spark : Spark’s core APIs
- Works on RDD level
- Works with python, R, Scala, Java

2. Layer 2 of Spark: Higher level APIs
- Works on dataframes, SQL, structed streaming, MLlib, GraphX

Sparkcontext vs Sparksession

Front door to Spark

• SparkContext is the entry point to Spark Core. It establishes the connection between a Spark application and the cluster, coordinates execution, and provides APIs for creating and manipulating RDDs
• SparkSession is main entry point to modern PySpark applications. It provides APIs to work with DataFrames, Spark SQL, and other Spark functionality, while also giving access to the underlying SparkContext for RDD-based operations

Basic PySpark

1. parallelize() is a method of SparkContext used to convert a local Python collection (like a list) into a distributed RDD.

data = [1, 2, 3, 4, 5, 6, 7, 8]

You can create an RDD with multiple partitions:
rdd = sc.parallelize(data, 4)

Conceptually:
RDD
│
├── Partition 1 → [1, 2]
├── Partition 2 → [3, 4]
├── Partition 3 → [5, 6]
└── Partition 4 → [7, 8]

2. Map: map() is a transformation used to apply a function to every element of an RDD and return a new RDD.

Get Saurabh Soni’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Basic syntax: rdd.map(lambda x: …)

result = rdd.map(lambda x: x * 2)
result = rdd.map(square)
result = rdd.map(lambda x: x.upper())

map() can change the structure

rdd = sc.parallelize(["John,25", "Mike,30", "Sarah,28"])
result = rdd.map(lambda x: x.split(","))
print(result.collect())
[['John', '25'], ['Mike', '30'], ['Sarah', '28']]

3. Reduce: reduce() combines/aggregates all elements of an RDD into a single value by repeatedly applying a function. reduce() is an Action

rdd.reduce(function)

from pyspark import SparkContext
from functools import reduce

sc = SparkContext.getOrCreate()

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

result = rdd.reduce(lambda x, y: x + y)

print(result)

Suppose the RDD has two partitions:

Partition 1: [1, 2, 3]
Partition 2: [4, 5]

Spark can first reduce within each partition

Partition 1 → 1 + 2 + 3 = 6
Partition 2 → 4 + 5     = 9

then combine
6 + 9 = 15

4. reduceByKey: reduceByKey() is an RDD transformation used to aggregate values that have the same key (groups values based on their keys). This map-side combining reduces the amount of data transferred during the shuffle.

rdd = sc.parallelize([
    ("apple", 10),
    ("banana", 20),
    ("apple", 15),
    ("banana", 5),
    ("orange", 30)
])

result = rdd.reduceByKey(lambda x, y: x + y)

print(result.collect())

[('apple', 25), ('banana', 25), ('orange', 30)]

Spark effectively does:

apple  → 10 + 15 = 25
banana → 20 + 5  = 25
orange → 30       = 30

5. filter: filter() is an RDD transformation used to keep only the elements that satisfy a given condition.

EXAMPLE 1:
rdd = sc.parallelize([1, 2, 3, 4, 5, 6])

result = rdd.filter(lambda x: x > 3)

print(result.collect())

[4, 5, 6]

EXAMPLE 2: Filtering key value RDDs
rdd = sc.parallelize([
    ("John", 5000),
    ("Mike", 3000),
    ("Sarah", 7000)
])

To keep employees earning more than 4,000:

result = rdd.filter(lambda x: x[1] > 4000)

print(result.collect())

[('John', 5000), ('Sarah', 7000)]

6. sortBy() allows you to specify any function/expression to determine the sorting key. It is a transformation

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

result = rdd.sortBy(lambda x: x, ascending=False)

print(result.collect())

[8, 5, 3, 2, 1]

Sort by a particular attribute:

rdd = sc.parallelize([
    ("John", 5000),
    ("Mike", 3000),
    ("Sarah", 7000)
])

result = rdd.sortBy(lambda x: x[1])

print(result.collect())

[
    ("Mike", 3000),
    ("John", 5000),
    ("Sarah", 7000)
]

7. sortByKey() is specifically for key-value RDDs. It is a transformation

rdd = sc.parallelize([
    (3, "C"),
    (1, "A"),
    (2, "B")
])

rdd.sortByKey().collect()

[(1, "A"), (2, "B"), (3, "C")]

8. distinct() is an RDD transformation used to remove duplicate elements from an RDD and return only unique values. Spark has to compare records across partitions, so distinct() generally involves a shuffle.

rdd = sc.parallelize([
    ("A", 10),
    ("B", 20),
    ("A", 10),
    ("C", 30)
])

rdd.distinct().collect()

[("A", 10), ("B", 20), ("C", 30)]

9. countByValue() is an RDD action that counts how many times each unique value occurs in an RDD. It works directly on individual values and don’t require creation of key value pairs
- countByValue() returns the result to the driver, whereas
- reduceByKey() returns another RDD that remains distributed.
That’s one reason reduceByKey() is more appropriate for large-scale/custom aggregations, while countByValue() is convenient for simple frequency counts when the number of distinct values is manageable.

countByValue() --> "Just count occurrences"
reduceByKey() --> "Aggregate values associated with each key"

rdd = sc.parallelize(["A", "B", "A", "C", "A"])

rdd.countByValue()

{
    'A': 3,
    'B': 1,
    'C': 1
}

while in reduceByKey:

rdd = sc.parallelize([
    ("A", 10),
    ("B", 20),
    ("A", 10),
    ("C", 30)
])

mapped_rdd = rdd.map(lambda x:(x,1))
results = mapped_rdd.reduceByKey(lambda x,y: x+y)
results.collect()

Let’s understand with an example:

words =
("big","Data","Is","SUPER","Interesting","BIG","data","IS","A","Trending","techn
ology")
## Converting to a distributed RDD
words_rdd = spark.sparkContext.parallelize(words)

## Using map function to normalising different cases of words to lower
words_normalized = words_rdd.map(lambda x:x.lower())

## Using map function to get key value pair with value 1 for each word as a key
mapped_words = words_normalized.map(lambda x:(x,1))

## get the count for each key (words in this case) 
aggregated_result = mapped_words.reduceByKey(lambda x,y:x+y)

## see the final results
aggregated_result.collect()

We can also chain all the commands and execute in one go

spark.sparkContext.parallelize(words).map(lambda x:x.lower()).map(lambda
x:(x,1)).reduceByKey(lambda x,y:x+y).collect()

## Cleaner way of writing the same - "\" means new line
result = spark. \
sparkContext. \
parallelize(words). \
map(lambda x:x.lower()). \
map(lambda x:(x,1)). \
reduceByKey(lambda x,y:x+y)

## Then
results.collect()

Check number of RDD Partitions:

<rdd>.getNumPartitions() is used to see number of partitions

- Consider there is file of 1 GB in HDFS, so # blocks would be 1 GB/128MB (default block size of 128MB) → 8 Blocks, so # of partitions will also be 8
# RDDpartitions = # Blocks in HDFS

- If the file size is less than 128 MB, then # blocks can be checked “spark.SparkContext.defaultMinPartitions” which is set as 2 by default
- “spark.SparkContext.defaultParallelism” is a property that indicates the # default tasks that can run in parallel