How to parse a huge file with csv data and calculate average on one of its column in plain Scala?

user_7119501 0
Viewed 174

I am trying to read a huge file (data in comma seperated values) The input file contains millions of rows in the below format. The file is also close to 8GB. Ex:

1, 22, Begin, session1
1, 33, End, session1
2, 20, Begin, session1
2, 30, End, session1
1, 30, Begin, session2
1, 50, End, session2
3, 90, Begin, session1
4, 10, Begin, session1
3, 100, End, session1
4, 20, End, session1
3, 200, OPEN, session2

The first value is the RECORDID, second value is its WEIGHT, the third value is the TRANSACTION_STATUS, fourth value is SESSIONID. I have to calculate the average of all WEIGHT of each RECORDID for all of its sessions between its BEGIN & END. If there is a session id without END, it should be ignored.

Example Output:

RECORD ID => 1, WEIGHTS => (33-22) = 11, (50-30)=20 => Average 15.5 
RECORD ID => 2, WEIGHTS => (30-20) = 10 => Average 10.0
RECORD ID => 3, WEIGHTS => (100-90) = 10 => Average 10.0
RECORD ID => 4, WEIGHTS => (20-10) = 10 => Average 10.0

Final output:

1, 15.5
2, 10.0
3, 10.0
4, 10.0

I started to code like below:

case class Users(recordid: Int, weight: Int, transaction_status: String, sessionid: String)
val userList = List[Users]()
val in = new BufferedReader(new InputStreamReader(new FileInputStream("/Users/Desktop/sessionfile.txt")))
Iterator continually in.readLine takeWhile (_ != null) foreach(println)

As my input file is a big file, I used InputStreamReader & Iterator to read records from the file. But I am a bit confused here because I previously did the above activity on a Spark Dataset where I created objects of my case class to represent the dataset with type Users => Dataset[Users] and Dataframe with spark SQL In this case, the code should only be written in plain Scala without using any implementation of Spark or SQL.

Could anyone let me know what is the efficient way to achieve the solution in plain scala code ? Any help is really appreciated.

scala
1 Answers

Here's one way to go about it.

val beginRE = raw"\s*(\d+)\s*,\s*(\d+)\s*,\s*Begin.*".r
val endRE   = raw"\s*(\d+)\s*,\s*(\d+)\s*,\s*End.*".r

util.Using(io.Source.fromFile("./inFile.csv")){
  _.getLines().foldLeft(Map[String,(Long,Int,Int)]()){
    case (acc, beginRE(recID, wght)) =>  //Begin record
      val (rt, cnt, _) = acc.getOrElse(recID,(0L,0,0))
      acc + (recID -> (rt, cnt, wght.toInt))
    case (acc, endRE(recID, wght)) =>    //End record
      val (rt, cnt, bgn) = acc.getOrElse(recID,(0L,0,-1))
      if (bgn < 0) {
        println(s"orphan End record: '$recID,$wght,...'")
        acc
      } else
        acc + (recID -> (rt + wght.toInt - bgn, cnt+1, -1))
    case (acc, rec) =>                   //bad record
      println(s"bad record: $rec")
      acc
  }
}.map(_.map{case (k,(rt,cnt,_)) => k -> rt/cnt.toDouble})
//res0: Try[Map[String,Double]] =
//    Success(Map(1 -> 15.5, 2 -> 10.0, 3 -> 10.0, 4 -> 10.0))

If there are multiple Begin records only the last one counts. The rest are ignored.

If there are multiple End records only the first one counts. The rest are reported as "orphaned."

user_49931280
Related