SPARK -DATAFRAME FILTERS

-
Consider we have a Students academic data as a CSV file and we need to filter the data of male students alone, i.e, we need to filter on Marks_obtained column. The Sample CSV file for our demo is as show below.


Problem is, Filter and fetch the records of students, whose MARKS_OBTAINED is more than or equal to 35.

Apply Filter using PySpark:

Filtering data from dataframe is a basic transformation in Apache Spark, which can be applied in different ways. 
  • Filter can be applied directly on a Spark dataframe using filter() API
  • we can use where() function on Spark dataframe 
  • we can also register the Spark dataframe as a temporary view and write a SQL query on top of it to  apply filter. 
We can also use multiple condition separated by and, or to apply filter conditions.

Syntax:

           df.filter("Filter condition")

Let us look into both the ways one by one. Open new Jupyter notebook and initiate Spark Session and Spark Context.

Coding:

Method 1: Using Temp view

Step 1: Read the input file as a dataframe.
Step 2: Register the dataframe as a temporary view using createOrReplaceTempView().
Step 3: Write a sql query and assign the output to dataframe as below.

Snippet;

df=spark.read.option('delimiter','|').csv('input.csv',header=True)                                       
                                                                                                                                                      
#Create temporary view on dataframe                                                                                    
df.createOrReplaceTempView('filter_view')                                                                           
                                                                                                                                                     
#Write a SQL query to filter and fetch record                                                                       
df_filter=spark.sql("""select * from filter_view where MARKS_OBTAINED >= 35""")


Method 2: Using Filter() or Where()

Step 1: Read the give raw file and create a Spark Dataframe to it.
Step 2: Use Filter() or Where() api, on top of Spark Dataframe. Refer the filter syntax mentioned in the above session.

The code snippet will be as follows,

df=spark.read.option('delimiter','|').csv('input.csv',header=True)
                                                                                                               
#apply filter api with filter condition                                                
df_filter=df.filter("marks_obtained >= 35")                                    


Output:

The filtered output is as shown below. I would recommend you to try this on your own to have a better understanding of concept.

out[]:

Full program:




Hope you enjoyed learning from this article. If you have any doubts or comments, please drop down in the below comment box.

Comments

Post a Comment

Popular posts from this blog

Spark interview question part-1

-
Image
  1. What is Apache Spark and what are the benefits of Spark over MapReduce? Spark is really fast. If run in-memory it is 100x faster than Hadoop MapReduce. In Hadoop MapReduce, you write many MapReduce jobs and then tie these jobs together using Oozie/shell script. This mechanism is very time consuming and MapReduce tasks have heavy latency. Between two consecutive MapReduce jobs, the data has to be written to HDFS and read from HDFS. This is time-consuming. In case of Spark, this is avoided using RDDs and utilizing memory (RAM). And quite often, translating the output of one MapReduce job into the input of another MapReduce job might require writing another code because Oozie may not suffice. In Spark, you can basically do everything from single code or console (PySpark or Scala console) and get the results immediately. Switching between ‘Running something on cluster’ and ‘doing something locally’ is fairly easy and straightforward. This also leads to less context switch of the d...
Read more

MongoDB - Data Modelling

-
Image
Data in MongoDB has a flexible schema.documents in the same collection. They do not need to have the same set of fields or structure Common fields in a collection’s documents may hold different types of data. Data Model Design MongoDB provides two types of data models: — Embedded data model and Normalized data model. Based on the requirement, you can use either of the models while preparing your document. Embedded Data Model In this model, you can have (embed) all the related data in a single document, it is also known as de-normalized data model. For example, assume we are getting the details of employees in three different documents namely, Personal_details, Contact and, Address, you can embed all the three documents in a single one as shown below − { _id : , Emp_ID : "10025AE336" Personal_details :{ First_Name : "Kishan" , Last_Name : "choudhary" , Date_Of_Birth : "1995-09-26" }, Contact : { e - mail : ...
Read more

SPARK - Deployment

-
Spark application, using spark-submit, is a shell command used to deploy the Spark application on a cluster. It uses all respective cluster managers through a uniform interface. Therefore, you do not have to configure your application for each one. Example Let us take the same example of word count, we used before, using shell commands. Here, we consider the same example as a spark application. Sample Input The following text is the input data and the file named is  in.txt . people are not as beautiful as they look, as they walk or as they talk. they are only as beautiful as they love, as they care as they share. Look at the following program − SparkWordCount.scala import org . apache . spark . SparkContext import org . apache . spark . SparkContext . _ import org . apache . spark . _ object SparkWordCount { def main ( args : Array [ String ]) { val sc = new SparkContext ( "local" , "Word Count" , "/usr/local/spark" , ...
Read more