Pyspark - Split large text file into multiple files

Viewed 3628

I need to split a large text file in S3 that can contain ~100 million records, into multiple files and save individual files back to S3 as .txt files. These records are not delimited and each column can be identified based on start and end positions. Length of each record varies based on "type" which is a string with a fixed start/end position and I need to split this file into multiple files based on value of "type".

E.g.

My name is Chris  age 45  
My name is Denni  age 46  
My name is Vicki  age 47  
My name is Denni  age 51  
My name is Chris  age 52

In the above example, assume my "record type" starts at 12th position and ends at 17th position. From a sequence of steps,

1. I need to get a distinct list of record types, which in this case are "Chris", "Denni" and "Vicki"

2. I need to split this file into 3 files, one for each record type and save them with same name as record types. Chris.txt, Denni.txt and Vicki.txt

Desired Output:

Chris.txt:

My name is Chris  age 45  
My name is Chris  age 52 

Denni.txt:

My name is Denni  age 46  
My name is Denni  age 51

Vicki.txt:

My name is Vicki  age 47 

I am using pyspark dataframes to achieve this and what I have now is something like this,

df_inter =df.select(df.value.substr(start,end).alias("Type"),df.value.alias("value"))

    df_types = df_inter.select("Type").distinct()
    type_count = df_types.count()

    while(i<type_count):
      type = df_types.select(df_types.Type).collect()[i][0]
      df_filtered = df_inter.filter(df_inter["Type"] == type)
      df_filtered.saveAsTextFile("path")
      i += 1

Current code works, but takes ~25 mins to process a 2.5 gb file with 5 node r5.xlarge EMR cluster, and takes much longer to process, say a 25 GB file. I would like to understand if there is a more efficient way of doing this and bring down the processing time. Appreciate your input.

1 Answers

I am assuming that your data is delimited by tab. you can load whole data into dataframe as shown below:

df = spark.read.format("com.databricks.spark.csv") \
  .option("mode", "DROPMALFORMED") \
  .option("header", "false") \
  .option("inferschema", "true") \
  .option("delimiter", '\t').load(PATH_TO_FILE)

+---+----+---+-----+---+---+
|_c0| _c1|_c2|  _c3|_c4|_c5|
+---+----+---+-----+---+---+
| My|name| is|Chris|age| 45|
| My|name| is|Denni|age| 46|
| My|name| is|Vicki|age| 47|
| My|name| is|Denni|age| 51|
| My|name| is|Chris|age| 52|
+---+----+---+-----+---+---+

from pyspark.sql.functions import col

Then you can filter the dataframe data and split into multiple dataframe depending on your column value.

Chris_df=df.filter(col('_c3')=='Chris')
+---+----+---+-----+---+---+
|_c0| _c1|_c2|  _c3|_c4|_c5|
+---+----+---+-----+---+---+
| My|name| is|Chris|age| 45|
| My|name| is|Chris|age| 52|
+---+----+---+-----+---+---+
Denni_df=df.filter(col('_c3')=='Denni')
+---+----+---+-----+---+---+
|_c0| _c1|_c2|  _c3|_c4|_c5|
+---+----+---+-----+---+---+
| My|name| is|Denni|age| 46|
| My|name| is|Denni|age| 51|
+---+----+---+-----+---+---+
Vicki_df=df.filter(col('_c3')=='Vicki')

+---+----+---+-----+---+---+
|_c0| _c1|_c2|  _c3|_c4|_c5|
+---+----+---+-----+---+---+
| My|name| is|Vicki|age| 47|
+---+----+---+-----+---+---+

I hope this work faster!

Related