Scala: speeding up word count on a file using (parallel) approach

user_926918 0
Viewed 265

I need to run word count on an input file of upto a million lines in scala. Each line is very long (>150K characters) too. The following is the standard program that works:

val wordCount = scala.io.Source.fromFile("sample.dat")
.getLines
.flatMap(_.split("\\W+"))
.foldLeft(Map.empty[String, Int]){
    (count, word) => count + (word -> (count.getOrElse(word, 0) + 1))
}

The below modification fails with the error, value par is not a member of Iterator[String]

val wordCount = scala.io.Source.fromFile("sample.dat")
.getLines
.flatMap(_.split("\\W+"))
.par
.foldLeft(Map.empty[String, Int]){
    (count, word) => count + (word -> (count.getOrElse(word, 0) + 1))
}

I am surprised with this as similar programs seem to work.

Further, I am wondering if par.reduce would be a faster and more efficient than a working par.foldLeft.

Would be grateful for any help or leads on this issue.

TIA

parallel-processing scala
1 Answers

Check this out. Parallel collections are gone from 2.13, though there is an external library, that you can use. Still, I say, if you are looking to process a large sequence of data in parallel, just use spark (especially, if you are going to need an external library anyway (you can run spark on single node ... and when you say, you need this solution "for testing" it feels weird that you want to test one solution, and then run a completely different one).

Here is a solution for you without external libraries (just for completeness):

    // First, create a local execution context to allow throttling the parallel jobs 
   val parallelism = 4 // how many chunks to process in parallel

   implicit val ec = ExecutionContext.fromExecutor(new ThreadPoolExecutor(
    parallelism, parallelism, 0L, TimeUnit.SECONDS,
    new ArrayBlockingQueue[Runnable](parallelism) {
      override def offer(e: Runnable) = {
        put(e)
        true
      }
    }
  ))

   // Now just split input into chunks and send to the executor
   // This does not read anything into memory yet
   
   val chunkSize = 4096 // how many lines to process at once

    val jobs = source 
     .getLines
     .grouped(chunkSize)
     .map { chunk => 
       Future {
          chunk
            .flatMap { _.split("""\W+""") }
            .foldLeft(Map.empty[String, Int]) { case (m, w) => 
                m + (w -> (m.getOrElse(w, 0) + 1))
            }
       }
     }

  // Now, combine the results.
  // This will fetch `parallelism*chunkSize` lines into memory and start
  // parallelism jobs processing the chunks. Once one of the jobs completes, 
  // it will read next `chunkSize` lines, and start another job. Etc.
    val result: Future[Map[String, Int]] = Future.sequence(jobs.toSeq).map { 
      _.reduce { (m1, m2) =>
        m1.foldLeft(m2){ case (m, (w, v)) => m + (w -> (m.getOrElse(w,0) + v)) }
      }
  }

The key to this is the ec implementation limiting the number of futures currently "in flight". You can wrap it, and the chunking logic into a small utility class, and make it reusable if you'd like. Though, I would still just use spark if I were you.

user_42545170
Related