how to handle the Exception in spark map() function?

Question

I want to ignore Exception in map() function , for example:

rdd.map(_.toInt)

where rdd is a RDD[String].

but if it meets non-number string, it will failed.

what is the easist way to ignore any Exception and ignore that line? (I do not want to use filter to handle exception, because there may be so many other exceptions...)

Answer 1

You can use a combination of Try and map/filter.

Try will wrap your computation into Success, if they behave as expected, or Failure, if an exception is thrown. Then you can filter what you want - in this case the successful computations, but you could also filter the error cases for logging purposes, for example.

The following code is a possible starting point. You can run and explore it in scastie.org to see if it fits your needs.

import scala.util.Try

object Main extends App {

  val in = List("1", "2", "3", "abc")

  val out1 = in.map(a => Try(a.toInt))
  val results = out1.filter(_.isSuccess).map(_.get)

  println(results)

}

Answer 2

I recommend you to use filter/map

rdd.filter(r=>NumberUtils.isNumber(r)).map(r=> r.toInt)

or flatmap

exampleRDD.flatMap(r=> {if (NumberUtils.isNumber(r)) Some(r.toInt) else  None})

Otherwise you can catch exception in map function

myRDD.map(r => { try{
        r.toInt
    }catch {
        case runtime: RuntimeException => {
        -1
        }
    }
})

and then apply filter(on -1)