[Python] Parse different schema

Need parsing approach to read a fixed length file efficiently

Sample File

A01DataHasBeenloaded

A02DataHasbeenParsedAndLoadedToMemory

A03CatIsInTheBag

A01....

A02....

A03....

and so on

The first group of A01,A02,A03 belongs to Record 1 and next group belongs to Record 2 and so on.One thing to note that the All lines staring with A01,A02 and A03 will have different schemas respectively.Currently the schema is as below

recordMarker: A

schema:
- column_name: Name_1

sequence: 'A01'

startchar: 1

length: 3

value: 'A01'(optional)


- column_name: Activity

sequence: 'A01'

startchar: 4

length: 14



- column_name: Name_2

sequence: 'A02'

startchar: 1

length: 3

value: 'A02'(optional key)


- column_name: Progress

sequence: 'A02'

startchar: 4

length: 16

- column_name: Name_3

sequence: 'A03'

startchar: 4

length: 37

value: 'A03'(optional key)

- column_name: Event

sequence: 'A03'

startchar: 4

length: 37


Sequence A01 has Name_1,Activity columns

Sequence A02 has Name_2,Progress columns

Sequence A03 has Name_3,Event columns

• 
  The sequence key actually helps in mapping which columns belongs to which schema.
  
  
• 
  The record marker is the prefix to filter out the records starting with A
  

I need to map each of A01,A02,A03 in continuation of lines to Single record(pivoting those schema).Resultant dataframe will have Name_1,Activity,Name_2,Event,Name_3,Progress columns.Currently I am using monotonically_increasing_id which could be inefficient in case, one of the schema(suppose with sequence A03) belonging to same record may reside in different partition and rest of the two(A01 and A02) in another partition which could lead to inconsistencies in record.

Current logic is as below:

def _create_sequence_dataframes(self, raw_df: DataFrame) -> Dict[str, DataFrame]:
sequence_dfs = {}
for field in self.record_schema.schema:
if field.sequence not in sequence_dfs:
sequence_df = raw_df.filter(
substring(col("value"), 1, len(self.record_marker) + 2)
== f"{self.record_marker}{field.sequence[-2:]}"
)
sequence_dfs[field.sequence] = sequence_df

sequence_dfs[field.sequence] = sequence_dfs[field.sequence].withColumn(
field.column_name,
substring(col("value"), field.startchar, field.length),
)

for sequence, df in sequence_dfs.items():
sequence_dfs[sequence] = df.drop("value")

return sequence_dfs


def _join_sequence_dataframes(
self, sequence_dfs: Dict[str, DataFrame]
) -> DataFrame:

sequences = sorted(list(sequence_dfs.keys()))

for seq in sequences:
sequence_dfs[seq] = sequence_dfs[seq].withColumn(
"row_id", monotonically_increasing_id()
)

base_df = sequence_dfs[sequences[0]]
for seq in sequences[1:]:
base_df = base_df.join(sequence_dfs[seq], "row_id", "outer")

result_df = base_df.drop("row_id")
return result_df


def process_data(
self, spark: SparkSession, file_path: str
) -> DataFrame:

raw_df = spark.read.text(file_path)
sequence_dfs = self._create_sequence_dataframes(raw_df)
joined_df = self._join_sequence_dataframes(sequence_dfs)
return joined_df


process_data is the main module which calls the other two.

Please suggest other approaches to parse the file efficiently

Continue reading...