Understanding the filter Function in PySpark

The filter function is one of the fundamental transformations you can apply to Resilient Distributed Datasets (RDDs) in PySpark. This function allows you to apply a function to each element of the RDD, resulting in a new RDD containing the transformed elements.

What is the filter Function in PySpark?

The filter function in PySpark is used to apply a transformation to each element of an RDD. The function takes a lambda function or a named function as an argument, processes each element, and returns a new RDD with the transformed data. This is particularly useful in scenarios where you need to perform operations such as data cleaning, conversion, or any element-wise transformation.

Using the filter Function in PySpark

Here’s how you can use the filter function in PySpark:

from pyspark.sql import SparkSession

spark_session = SparkSession.builder.master("local").appName("simple filter example Pyspark").getOrCreate();

data = [1, 2, 3, 4, 5]
rdd = spark_session.sparkContext.parallelize(data)
squared_rdd = rdd.filter(lambda ele: ele%2==0)
print(squared_rdd.collect())

Try it Yourself

Passing a function to PySpark filter - RDD

Here’s how you can use the filter function in PySpark:

from pyspark.sql import SparkSession
 
spark_session = SparkSession.builder.master("local").appName("simple filter example Pyspark").getOrCreate();

def square(ele):
val = ele%2!=0
return val

data = [1, 2, 3, 4, 5]
rdd = spark_session.sparkContext.parallelize(data)
squared_rdd = rdd.filter(square)
print(squared_rdd.collect())

Real-World Example: Processing Log file

Let’s look at the below log file , we need to extract and then filter IP address from it. Please click here to download data and try this on your machine.

from pyspark.sql import SparkSession
spark_session=SparkSession.builder.master("local").appName("process log file using pyspark RDD").getOrCreate();

raw_rdd=spark_session.sparkContext.textFile("/Users/apple/PycharmProjects/pyspark/data/text/log.txt")
rdd=raw_rdd.map( lambdaline: line.split('-')[0] )
ip_rdd=rdd.filter(lambda ip_address: '38.99.236.50' in ip_address)
for line in ip_rdd.collect():
print(line)

Output:

Performance Considerations

While the filter function is powerful, it’s important to be mindful of the potential performance implications when working with large datasets. Since the filter function executes the transformation on each element of the RDD, ensuring that the transformation function is efficient is crucial for minimizing processing time.

Conclusion

The filter function is an essential tool in the PySpark arsenal, enabling developers to apply transformations across RDDs with ease. By understanding how to use filter effectively, you can unlock powerful data processing capabilities in your PySpark applications.