Most Important PySpark Functions with Example

Let’s see the practical Implementations:-

Example:- A.) Concatenating one or more columns using expr()

# concate friend's name, age, and location columns using expr()
df_concat = df_friends.withColumn("name-age-location",expr("friends_name|| '-'|| age || '-' || location"))
df_concat.show()

We have joined the name, age, and location columns and stored the result in a new column called “name-age-location.”

Example:- B.) Add a new column based on a condition (CASE WHEN) using expr() 

# check if exercise needed based on weight
# if weight is more or equal to 60 -- Yes
# if weight is less than 55 -- No
# else -- Enjoy
df_condition = df_friends.withColumn("Exercise_Need", expr("CASE WHEN weight >= 60 THEN 'Yes' " + "WHEN weight < 55 THEN 'No' ELSE 'Enjoy' END"))
df_condition.show()

Our “Exercise_Need” column received three values (Enjoy, No, and Yes) based on the condition given in CASE WHEN. The first value of the weight column is 58, so it’s less than 60 and more than 55, so the result is “Enjoy.”

Example:- C.) Creating a new column using the current column value inside the expression.

# let increment the meetup month by the number of offset
df_meetup = df_friends.withColumn("new_meetup_date", expr("add_months(meetup_date,offset)"))
df_meetup.show()

The “meetup_date” month value increases by the offset value, and the newly generated result is stored in the “new_meetup_date” column.

The Padding Functions

A.) lpad():- 

This function provides padding to the left side of the column, and the inputs for this function are column name, length, and padding string.

B.) rpad ():- 

This function is used to add padding to the right side of the column. Column name, length, and padding string are additional inputs for this function.

Note:- 

• If the column value is longer than the specified length, the return value will be shortened to length characters or bytes.
• If the padding value is not specified, then the column value will be padded to the left or right depending on the function you are using, with space characters if it is a character string and with zeros if it is a byte sequence.

Let’s first create a data Frame:-

# importing necessary libs
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lpad, rpad

# creating session
spark = SparkSession.builder.appName("practice").getOrCreate()

# creating data
data = [("Delhi",30000),("Mumbai",50000),("Gujrat",80000)]
columns= ["state_name","state_population"]
df_states = spark.createDataFrame(data = data, schema = columns)
df_states.show()

Example:- 01 – Use of left padding

# left padding
df_states = df_states.withColumn('states_name_leftpad', lpad(col("state_name"), 10, '#'))
df_states.show(truncate =False)

We added the ‘#’ symbol to the left of the “state_name” column values, and the total length of column values becomes “10″ after the padding.

Example:-02 – Right padding

# right padding
df_states = df_states.withColumn('states_name_rightpad', rpad(col("state_name"), 10, '#'))
df_states.show(truncate =False)

We added the “#” symbol to the right of the “state_name” column values, and the total length becomes ten after the right padding.

Example:-03 – When the column string length is longer than the padded string length

df_states = df_states.withColumn('states_name_condition', lpad(col("state_name"), 3, '#'))
df_states.show(truncate =False)

In this case, the return column value will be shortened to the length of the padded string length. You can see the “state_name_condition” column only has values of length 3, which is the padded length we have given in the function.

The repeat() Function

In PySpark, we use the repeat function to duplicate the column values. The repeat(str,n) function returns the string containing the specified string value repeated n times.

Example:- 01 

# importing necessary libs
from pyspark.sql import SparkSession
from pyspark.sql.functions import expr, repeat

# creating session
spark = SparkSession.builder.appName("practice").getOrCreate()

# # create data
data = [("Prashant",25, 80), ("Ankit",26, 90),("Ramakant", 24, 85)]
columns= ["student_name", "student_age", "student_score"]
df_students = spark.createDataFrame(data = data, schema = columns)
df_students.show()

# repeating the column (student_name) twice and saving results in new column
df_repeated = df_students.withColumn("student_name_repeated",(expr("repeat(student_name, 2)")))
df_repeated.show()

We have repeated the “student_name” column values in the above example twice.

We can also use this function with the Concat function, where we can repeat some string values n times before column values, working like padding, where n may be the length of some values.

The startswith() and endswith() function

startswith():-

It will produce a boolean result of True or False. When the Dataframe column value ends with the string provided as a parameter to this method, it returns True. If no match is found, it returns False.

endswith():-

The boolean value (True/False) will be returned. When the DataFrame column value ends with a string supplied as an input to this method, it returns True. False is returned if not matched.

Note:-

• Return 𝐍𝐔𝐋𝐋 if either of the column values or input strings are 𝐍𝐔𝐋𝐋.
• Return 𝗧𝗿𝘂𝗲 if the input check strings are empty.
• These methods are case-sensitive.

Create a data frame:-

# importing necessary libs
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# creating session
spark = SparkSession.builder.appName("practice").getOrCreate()

# # create dataframe
data = [("Prashant",25, 80), ("Ankit",26, 90),("Ramakant", 24, 85), (None, 23, 87)]
columns= ["student_name", "student_age", "student_score"]
df_students = spark.createDataFrame(data = data, schema = columns)
df_students.show()

Example – 01  First, check the output type.

df_internal_res = df_students.select(col("student_name").endswith("it").alias("internal_bool_val"))
df_internal_res.show()

• The output is a boolean value.
• The output value is null for the last row value because the corresponding value of the “students_name” column is NULL.

Example – 02

• Now we use the filter() method to fetch the rows corresponding to the True values.

df_check_start = df_students.filter(col("student_name").startswith("Pra"))
df_check_start.show()