Explore topic-wise InterviewSolutions in .

If any RDD partition is lost, then that partition can be recomputed using operations lineage from the original fault-tolerant dataset.

PySpark has GAINED immense popularity among the big data and machine learning enthusiasts as well as in the organizations like NETFLIX as it exposes Spark functionalities in python that helps developers collaborate with powerful libraries that aids in big data PROCESSING. In this article, we have seen the most commonly asked INTERVIEW questions in PySpark that would help developers crack interviews.

References

• PySpark by Examples
• PySpark Documentation 
• Python

PySpark STREAMING is scalable, fault-tolerant, HIGH throughput based processing streaming system that supports streaming as well as batch loads for supporting real-time data from data sources like TCP Socket, S3, Kafka, Twitter, file system folders etc. The processed data can be sent to live dashboards, Kafka, databases, HDFS etc.

To perform Streaming from the TCP socket, we can use the readStream.format("socket") method of Spark session object for READING data from TCP socket and providing the streaming source host and port as options as shown in the code below:

Spark loads the data from the socket and represents it in the value column of the DataFrame object. The df.printSchema() prints

Post data processing, the DataFrame can be streamed to the console or any other destinations based on the REQUIREMENTS like Kafka, dashboards, database etc.

We can make use of the join() method present in PySpark SQL. The syntax for the function is:

where,
other - Right side of the join
on - column name string used for joining
how - type of join, by default it is INNER. The values can be inner, left, right, cross, full, outer, left_outer, right_outer, left_anti, left_semi.

The join expression can be appended with where() and filter() methods for filtering ROWS. We can have multiple join too by means of the chaining join() method.

Consider we have two dataframes - employee and department as shown below:

We can inner join the Employee DataFrame with Department DataFrame to get the department information along with employee information as:

The result of this becomes:

We can also perform joins by chaining join() method by following the syntax:

Consider we have a third dataframe called Address DataFrame having columns emp_id, city and state where emp_id acts as the foreign key equivalent of SQL to the Employee DataFrame as shown below:

If we want to get address details of the address along with the Employee and the Department Dataframe, then we can run,

The resultDf would be:

PySpark SQL is the most popular PySpark MODULE that is used to process structured columnar data. Once a DataFrame is CREATED, we can interact with data using the SQL syntax. Spark SQL is used for bringing native raw SQL queries on Spark by using select, where, group by, join, union etc. For using PySpark SQL, the first step is to create a temporary table on DataFrame by using createOrReplaceTempView() function. Post creation, the table is accessible throughout SparkSession by using sql() method. When the SparkSession GETS terminated, the temporary table will be dropped.
For example, consider we have the following DataFrame assigned to a variable df:

In the below piece of code, we will be creating a temporary table of the DataFrame that gets accessible in the SparkSession using the sql() method. The SQL queries can be run within the method.

The schema will be displayed as shown below:

For the above example, let’s try running group by on the Gender column:

The above statements RESULTS in:

These methods belong to the Column class and are used for searching DataFrame ROWS by CHECKING if the column value starts with some value or ends with some value. They are used for filtering data in applications.

• startsWith() – returns boolean Boolean value. It is true when the value of the column starts with the specified string and False when the match is not satisfied in that column value.
• endsWith() – returns boolean Boolean value. It is true when the value of the column ends with the specified string and False when the match is not satisfied in that column value.

Both the methods are case-sensitive.

Consider an example of the startsWith() method here. We have created a DataFrame with 3 rows:

If we have the below code that checks for returning the rows where all the names in the Name column start with “H”,

The output of the code would be:

Notice how the record with the Name “Ron” is filtered out because it does not start with “H”.

Yes, it is! Realtime applications make use of external file systems like LOCAL, HDFS, HBase, MySQL table, S3 Azure etc. Following example shows how we can create DATAFRAME by reading data from a csv file present in the local system:

PySpark supports csv, TEXT, AVRO, parquet, tsv and many other file extensions.

We can do it by MAKING USE of the createDataFrame() method of the SparkSession.

This creates the dataframe as shown below:

We can GET the schema of the dataframe by using df.printSchema()

The FOLLOWING image represents how we can visualize RDD creation in PySpark:

In the image, we see that the data we have is the list form and post converting to RDDs, we have it stored in different partitions.
We have the following approaches for CREATING PySpark RDD:

• USING sparkContext.parallelize(): The parallelize() method of the SparkContext can be used for creating RDDs. This method loads EXISTING collection from the driver and parallelizes it. This is a basic approach to create RDD and is used when we have data already present in the memory. This also requires the presence of all data on the Driver before creating RDD. Code to create RDD using the parallelize method for the python list shown in the image above:

• Using sparkContext.textFile(): Using this method, we can read .txt file and convert them into RDD. Syntax:

• Using sparkContext.wholeTextFiles(): This function returns PairRDD (RDD CONTAINING key-value pairs) with file path being the key and the file content is the value.

We can also read csv, json, parquet and various other formats and create the RDDs.

• Empty RDD with no partition using sparkContext.emptyRDD: RDD with no data is called empty RDD. We can create such RDDs having no partitions by using emptyRDD() method as shown in the code piece below:

• Empty RDD with partitions using sparkContext.parallelize: When we do not require data but we require partition, then we create empty RDD by using the parallelize method as shown below:

To create SparkSession, we use the builder pattern. The SparkSession class from the pyspark.sql library has the getOrCreate() method which creates a new SparkSession if there is none or else it returns the existing SparkSession object. The following code is an example for creating SparkSession:

Here,

• master() – This is used for setting up the mode in which the application has to run - cluster mode (use the master NAME) or standalone mode. For Standalone mode, we use the local[x] value to the FUNCTION, where x represents partition count to be created in RDD, DataFrame and DATASET. The value of x is ideally the number of CPU cores available.
• appName() - Used for setting the application name
• getOrCreate() – For returning SparkSession object. This creates a new object if it does not exist. If an object is there, it simply returns that.

If we want to create a new SparkSession object every TIME, we can use the newSession method as SHOWN below:

Custom PROFILERS are supported in PYSPARK. These are useful for building predictive models. Profilers are useful for data REVIEW to ensure that it is valid and can be used for consumption. When we require a custom PROFILER, it has to define some of the following methods:

• profile: This PRODUCES a system profile of some sort.
• stats: This returns collected stats of profiling.
• dump: This dumps the profiles to a specified path.
• add: This helps to add profile to existing accumulated profile. The profile class has to be selected at the time of SparkContext creation.
• dump(id, path): This dumps a specific RDD id to the path given.

Consider an example where we want to capitalize the first letter of every word in a string. This feature is not supported in PySpark. We can however achieve this by creating a UDF capitalizeWord(str) and using it on the DataFrames. The following STEPS demonstrate this:

• Create Python function capitalizeWord that takes a string as input and capitalizes the first CHARACTER of every word.

• Register the function as a PySpark UDF by using the udf() method of org.apache.spark.sql.functions.udf package which needs to be imported. This method returns the object of class org.apache.spark.sql.expressions.UserDefinedFunction.

• USE UDF with DataFrame: The UDF can be applied on a Python DataFrame as that acts as the built-in function of DataFrame.
  Consider we have a DataFrame of stored in variable df as below:

To capitalize every first character of the word, we can use:

The output of the above code would be:

UDFs have to be designed in a way that the algorithms are efficient and take less time and space complexity. If care is not taken, the performance of the DataFrame OPERATIONS would be impacted.

PySpark SparkConf is used for setting the configurations and parameters required to RUN applications on a cluster or local system. The following class can be executed to run the SparkConf:

where:

• loadDefaults - is of type boolean and indicates whether we require loading values from Java System PROPERTIES. It is True by default.
• _jvm - This belongs to the class py4j.java_gateway.JVMView and is an internal parameter that is used for passing the handle to JVM. This need not be set by the users.
• _jconf - This belongs to the class py4j.java_gateway.JavaObject. This parameter is an OPTION and can be used for passing existing SparkConf handles for USING the parameters.

The most common workflow followed by the spark program is:

• The first STEP is to create input RDDs depending on the external data. Data can be obtained from different data sources.
• Post RDD creation, the RDD transformation operations like FILTER() or map() are run for creating new RDDs depending on the BUSINESS logic.
• If any intermediate RDDs are required to be REUSED for later purposes, we can persist those RDDs.
• Lastly, if any action operations like first(), count() etc are present then spark launches it to initiate parallel computation.

DAG stands for Direct ACYCLIC Graph. DAGScheduler constitutes the scheduling layer of Spark which implements scheduling of tasks in a stage-oriented manner using jobs and stages. The logical execution plan (Dependencies lineage of transformation actions upon RDDs) is transformed into a physical execution plan CONSISTING of stages. It computes a DAG of stages needed for each job and keeps track of what stages are RDDs are materialized and finds a minimal schedule for running the jobs. These stages are then submitted to TaskScheduler for running the stages. This is represented in the image flow below:

DAGScheduler performs the following three things in Spark:

• Compute DAG execution for the job.
• Determine preferred locations for running each task
• Failure Handling due to OUTPUT files lost during shuffling.

PySpark’s DAGScheduler follows event-queue architecture. Here a thread POSTS events of type DAGSchedulerEvent such as new stage or job. The DAGScheduler then reads the stages and sequentially EXECUTES them in topological order.

PySpark SIMILAR to Apache Spark works in master-slave architecture PATTERN. Here, the master node is called the Driver and the slave nodes are called the workers. When a Spark application is run, the Spark Driver creates SparkContext which acts as an entry point to the spark application. All the OPERATIONS are executed on the worker nodes. The RESOURCES required for executing the operations on the worker nodes are managed by the Cluster Managers. The following diagram illustrates the architecture described: