Process txt file with Spark

Question

I need to read text file into dataset[T] in Spark. The file is not properly formatted, as it has some blank fields and it is hard to define the parameter to split a string even. I have been trying to read the data into RDD and then transform it into case class type, however, not all the fields are parsed properly and I get an error:

java.lang.NumberFormatException: empty String
        at sun.misc.FloatingDecimal.readJavaFormatString(FloatingDecimal.java:1842)
        at sun.misc.FloatingDecimal.parseDouble(FloatingDecimal.java:110)
        at java.lang.Double.parseDouble(Double.java:538)
        at scala.collection.immutable.StringLike.toDouble(StringLike.scala:321)
        at scala.collection.immutable.StringLike.toDouble$(StringLike.scala:321)
        at scala.collection.immutable.StringOps.toDouble(StringOps.scala:33)
        at captify.test.spark.Stats$$anonfun$2.apply(Stats.scala:53)
        at captify.test.spark.Stats$$anonfun$2.apply(Stats.scala:53)
        at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
        at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
        at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$2.hasNext(WholeStageCodegenExec.scala:636)
        at org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:255)
        at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:836)
        at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:836)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
        at org.apache.spark.scheduler.Task.run(Task.scala:121)
        at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:411)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

What can I do to process this file properly? My .txt file looks like this (anonimized random data but the format is the same):

NEW50752085  84.0485 -76.3851  85.1   THE NAME OF AN OBJECT                       
DEM00752631  51.9581 -85.3315  98.5   THE NAME OF AN OBJECT                                  
KI004867205  40.8518  15.9351 276.5   THE NAME OF AN OBJECT           FHG   41196

I tried to process it in such a way:

    val dataRdd = spark.sparkContext
      .textFile("file.txt")

    val dataArray = dataRdd
      .map(_.split(" "))

  case class caseClass(
    c1: String,
    c2: Double,
    c3: Double,
    c4: Double,
    c5: String,
    c6: String,
    c7: String
  )

    val df = dataArray
      .map(record => (record(0), record(1).toDouble, record(2).toDouble, record(3).toDouble, record(4), record(5), record(6)))
      .map{case (c1, c2, c3, c4, c5, c6, c7) => CaseClass(c1, c2, c3, c4, c5, c6, c7)
      }.toDF()

user_11498964 · Answer

I'm going to be making some assumptions in this answer that might be incorrect but I believe them to be true based on your data given and errors provided.

Assumption 1: Your data is space-delimited, by several spaces. I came to this assumption based on your NumberFormatException of empty strings you provided. If your file was delimited by tabs, we would not have run into this.
Assumption 2 (This is for my own sake of mind, but might not be true): Each data element is space-delimited by the same number of spaces. For the remainder of this answer I will assume that number of spaces is four. If this assumption is not the case, this becomes a far more difficult problem.
Assumption 3: Only the last 2 of 7 data elements are optional, and sometimes do not appear.

Your NumberFormatException is caused by you splitting by one space. Assume the following row is delimited by spaces:

NEW50752085    84.0485    -76.3851    85.1    THE NAME OF AN OBJECT

When you split on one space, this row gets converted into the following Array:

Array(NEW50752085, "", "", "", 84.0485, "", "", "", -76.3851, "", "", "", 85.1, "", "", "", THE, NAME, OF, AN, OBJECT)

The second element of this array, which is an empty string, is what you are trying to convert into a Double. That is what gives you the NumberFormatException on empty string.

.map(_.split("    "))

When you change this to be split on 4 spaces (based on my assumption, which may or may not be true) you get the following:

Array(NEW50752085, 84.0485, -76.3851, 85.1, THE NAME OF AN OBJECT)

But now we run into another problem - this only has five elements! We want seven.

We can change this by modifying your later code:

val df = dataArray.map(record => {
  (record(0), record(1).toDouble, record(2).toDouble, record(3).toDouble, record(4), 
  if(record.size > 5) record(5) else "",
  if(record.size > 6) record(6) else "")
}).map{case (c1, c2, c3, c4, c5, c6, c7) => caseClass(c1, c2, c3, c4, c5, c6, c7)}.toDF

df.show
+-----------+-------+--------+----+--------------------+---+-----+
|         c1|     c2|      c3|  c4|                  c5| c6|   c7|
+-----------+-------+--------+----+--------------------+---+-----+
|NEW50752085|84.0485|-76.3851|85.1|THE NAME OF AN OB...|   |     |
|DEM00752631|51.9581|-85.3315|98.5|THE NAME OF AN OB...|   |     |
|KI004867205|40.8518| 15.9351|76.5|THE NAME OF AN OB...|FHG|41196|
+-----------+-------+--------+----+--------------------+---+-----+

Again, this approach will only work if all elements are delimited by the same number of spaces.

user_3741859 · Answer

If your data doesn't have a closed format readable by spark the only option you have is to create your own custom reader using FileInputFormat

This way you will be able to define a resolution flow to each line of your data determine how in split and deal with edge cases.

The best way to dive into it is by example. This is a quite solid one: https://www.ae.be/blog-en/ingesting-data-spark-using-custom-hadoop-fileinputformat/

Process txt file with Spark

2 Answers