I have the following PySpark code to loop through several subfolders and process the same code on the data within the subfolders, then write to different output folders. input_folders = ["/root_folder/subfolder_1","/root_folder/subfolder_2","/root_folder/subfolder_3"] index=0 for folder in input_folders: df = spark.read.parquet(folder) df_1 = df.repartition(24, "colA", "colB") e_rdd = df_1.rdd.mapPartitions(e_partition) clean_e_rdd = e_rdd.filter(lambda row: len(row) == 30) e_df = spark.createDataFrame(clean_e_rdd, output_schema) df_filter = e_df.filter((e_df["colA"]=="A") & (e_df["colB"]=="B")) df_filter.select(*columns_to_write).write.mode("append").parquet(output_path) index += 1 e_partition is a function e_partition(iterator) that returns rows of the df_1 dataframe, along with 6 additional columns after performing some processing on certain columns of df_1. My problem is I'm getting an empty dataframe for the results of subfolder_1, which is probably correct. However, I'm also getting empty dataframes as results of subfolder_2 and subfolder_3, which is wrong. When I ran the code only for subfolder_2, I'm getting non-empty dataframes as results. What is wrong with my code, and how can I resolve this? Continue reading...
