<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Example dataset
Step1:  Select Rows by index
First, we need to add index to each rows. The zipWithIndex function zips the RDD elements with their corresponding index and returns the result as a new element.
Step2:  Now we can apply the map function to modify the structure of each element. Assume x is an element from the above RDD object, x has two elements
Step3:  Let's add column names and save the result.
Step4:  After we obtain the index column, we can apply the pyspark.sql.DataFrame.filter function to select rows of the DataFrame. The filter function takes a column of types.BooleanType as input.
Select specific rows
Step5:  Select rows between a range
Step6:  Select rows by a cutoff index
Step7:  Select rows by logical criteria
Example 1
Step8:  Example 2
Step9:  <span style="color
Step10:  Now we need to add the new column to the original DataFrame. This can be done by applying the select() function to select all original columns as well as the new filtering columns.
Step11:  Now we can filter the DataFrame by the requested conditions. After we filter the DataFrame, we can drop the filtering column.
<ASSISTANT_TASK:>
Python Code:
mtcars = spark.read.csv('../../data/mtcars.csv', inferSchema=True, header=True)
# correct first column name
mtcars = mtcars.withColumnRenamed('_c0', 'model')
mtcars.show(5)
Explanation: Example dataset
End of explanation
mtcars.rdd.zipWithIndex().take(5)
Explanation: Select Rows by index
First, we need to add index to each rows. The zipWithIndex function zips the RDD elements with their corresponding index and returns the result as a new element.
End of explanation
mtcars.rdd.zipWithIndex().map(lambda x: [x[1]] + list(x[0])).take(5)
Explanation: Now we can apply the map function to modify the structure of each element. Assume x is an element from the above RDD object, x has two elements: x[0] and x[1]. x[0] is an Row object, and x[1] is the index, which is an integer. We want to merge these two values to create a list. And we also want the first element in the list is the index.
End of explanation
header = ['index'] + mtcars.columns
mtcars_df = mtcars.rdd.zipWithIndex().map(lambda x: [x[1]] + list(x[0])).toDF(header)
mtcars_df.show(5)
Explanation: Let's add column names and save the result.
End of explanation
mtcars_df.filter(mtcars_df.index.isin([1,2,4,6,9])).show()
Explanation: After we obtain the index column, we can apply the pyspark.sql.DataFrame.filter function to select rows of the DataFrame. The filter function takes a column of types.BooleanType as input.
Select specific rows
End of explanation
mtcars_df.filter(mtcars_df.index.between(5, 10)).show()
Explanation: Select rows between a range
End of explanation
mtcars_df.filter(mtcars_df.index < 9).show()
mtcars_df.filter(mtcars_df.index >= 14).show()
Explanation: Select rows by a cutoff index
End of explanation
mtcars_df.filter(mtcars_df.cyl == 4).show()
Explanation: Select rows by logical criteria
Example 1: select rows when cyl = 4
End of explanation
from pyspark.sql import functions as F
Explanation: Example 2: select rows when vs = 1 and am = 1
When the filtering is based on multiple conditions (e.g., vs = 1 and am = 1), we use the conditions to build a new boolean type column. And we filter the DataFrame by the new column.
End of explanation
filtering_column = F.when((mtcars_df.vs == 1) & (mtcars_df.am == 1), 1).name('filter_col')
filtering_column
Explanation: <span style="color:red">Warning: when passing multiple conditions to the when() function, each condition has to be within a pair of parentheses</span>
End of explanation
all_original_columns = [eval('mtcars_df.' + c) for c in mtcars_df.columns]
all_original_columns
all_columns = all_original_columns + [filtering_column]
all_columns
new_mtcars_df = mtcars_df.select(all_columns)
new_mtcars_df.show()
Explanation: Now we need to add the new column to the original DataFrame. This can be done by applying the select() function to select all original columns as well as the new filtering columns.
End of explanation
new_mtcars_df.filter(new_mtcars_df.filter_col == 1).drop('filter_col').show()
Explanation: Now we can filter the DataFrame by the requested conditions. After we filter the DataFrame, we can drop the filtering column.
End of explanation
<END_TASK>