Apache Spark bad record for large number of columns having pattern in name

user_14921789 0
Viewed 145

I am new to Spark and was trying to handle bad records. I found the below while searching

val df = spark.read
          .option("badRecordsPath", "/tmp/badRecordsPath")
          .schema("Country String, Rank Integer")
          .csv("/tmp/inputFile.csv")

We are reading multiple files and need to validate the schema. But the problem here is that in each input CSV file we have some column names with a specific pattern and it can be dynamic in numbers in each file. For example, consider the below headers in the first CSV file

ID, POL_1, POL_2, POL_3, . . .  . POL500

The second CSV file have below headers

ID, POL_1, POL_2, POL_3, ...,POL_10

Now for each file we have a different number of headers and with a similar pattern. So in that case I can not hard code the schema String or Struct fields. Is there any way to achieve the same?

One solution I can think of preload the input file each time and extract schema and then apply regex and assign the data type. But in this case, we have to load twice the file.

Can anyone please help me with this because I have been struggling with this.

apache-spark-sql apache-spark java
1 Answers

I had a similar case, however, I had a finite number of types of files, I do not know if this is your case, what I did:

  1. Read the first line of the CSV file.
  2. Case/Switch on the first line to pick the right schéma
  3. Ingest files.
  4. Union by name all the dataframe (my goal was to have a large dataframe at the end).

A kinda similar example can be found here: https://github.com/jgperrin/net.jgp.books.spark.ch15/blob/master/src/main/java/net/jgp/books/spark/ch15/lab300_nyc_school_stats/NewYorkSchoolStatisticsApp.java

If your header is completely variable, then you can create a schema dynamically based on reading the first line of your CSV and extrapolating the schéma from it. It should not be that difficult but you will need to make sure that your options are the same for both your schema builder and your actual read(). I would isolate that in a utility function that does both the schema building and ingestion.

Another option is to load everything without schema and then validate the data after as part of your DQ normalization. It kind be painful, but it can also give you better control on what your looking for as anomalies.

user_31972440
Related