Create RDD

PySpark's SparkContext class provides the below methods to create an RDD. Let's see each with an example.

• parallelize()
• emptyRDD()
• textFile()
• wholeTextFiles()

Create RDD from PySpark Shell or REPL

You have learned in the previous lesson that the PySpark shell provides a sparkContext variable named “sc”, hence, you can use sc.parallelize() to create an RDD. The below example creates an RDD with a list of integers.

>>> rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
>>> print(rdd.collect())
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

Creating RDD from the Application

You can create RDDs in PySpark from various data sources, such as an existing collection, HDFS (Hadoop Distributed File System), local files, S3, and more. RDDs are created primarily in two different ways: first, by parallelizing an existing collection; and secondly, by referencing a dataset from an external storage system like a local file system, HDFS, S3, and many more.

Before we create an RDD, we need to initialize a SparkSession object using the builder pattern by passing the master and appName details as shown below.

# Create SparkSession
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[1]").appName("AIStudyLabs.com").getOrCreate()

Using parallelize()

sparkContext.parallelize() is used to parallelize an existing collection in your driver program. This is a basic method to create RDDs and is used mainly during POCs or prototyping, as it requires all data to be present on the driver program prior, hence it is not most used for production applications. However, it is widely used for testing purposes.

# Create RDD from parallelize    
data_seq = [("Java", 20000), ("Python", 100000), ("Scala", 3000)]
rdd_parallelize = spark.sparkContext.parallelize(data_seq)

>>> print(rdd_parallelize.collect())
[('Java', 20000), ('Python', 100000), ('Scala', 3000)]

Create Empty RDD

Sometimes we may need to create an empty RDD, and you can use either emptyRDD() method or parallelize().

# Creates an empty RDD with no partition
empty_rdd = spark.sparkContext.emptyRDD()  # creates EmptyRDD
empty_rdd2 = spark.sparkContext.parallelize([])  # also creates EmptyRDD

# Print
print("Is Empty RDD: ", empty_rdd.isEmpty())

# Output: Is Empty RDD  :  True