[Python] How to Read Multiple CSV Files with Skipping Rows and Footer in PySpark Efficiently?

I have several CSV files with an inconsistent number of data rows without a header row and I want to read these files into a single PySpark DataFrame. The structure of the CSV files is as follows:

data1,data2
data1,data2,data3
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 => data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6 data1,data2,data3,data4,data5,data6
data1,data2
data1,data2
data1,data2,data3


I want to skip the first two rows and the last three rows from each file before combining them into one DataFrame. I am using the following approach, which works but I am looking for a more optimized solution because the datasets is quite large.

def concat(df_list: list):
df = df_list[0]
for i in df_list[1:]:
df = df.unionByName(i, allowMissingColumns=True)
return df


def __read_with_separators(self, spark: SparkSession, field_details: List[Dict[str, Any]], file_path_list: List[str], kwargs: dict) -> DataFrame:
df_list = []
for file_path in file_path_list:
rdd = spark.sparkContext.textFile(file_path)

total_rows = rdd.count()
start_index = kwargs.get("skiprows", 0)
end_index = total_rows - kwargs.get("skipfooter", 0)

rdd_filtered = rdd.zipWithIndex().filter(lambda x: start_index <= x[1] < end_index).map(lambda x: x[0]).map(lambda line: line.split(delimiter))
temp_df = rdd_filtered.toDF(schema)
df_list.append(temp_df)

return concat(df_list)


Questions:

1. Is there a way to read multiple CSV files at once while skipping rows and footer rows more efficiently?
2. Are there any optimizations or improvements to the current approach to handle large datasets more effectively?

Continue reading...