How to test exceptions thrown inside map in Scala

user_1808602 0
Viewed 346

I have the following Scala function

def throwError(spark: SparkSession,df:DataFrame): Unit = {
        import spark.implicits._
        throw new IllegalArgumentException(s"Illegal arguments")
        val predictionAndLabels = df.select("prediction", "label").map {
            case Row(prediction: Double, label: Double) => (prediction, label)
            case other => throw new IllegalArgumentException(s"Illegal arguments")
        }
        predictionAndLabels.show()
}

I want to test exceptions thrown by the above function, but my tests fail.

"Testing" should "throw error for datetype" in withSparkSession {
    spark => {

     // Creating a dataframe 
      val someData = Seq(
        Row(8, Date.valueOf("2016-09-30")),
        Row(9, Date.valueOf("2017-09-30")),
        Row(10, Date.valueOf("2018-09-30"))
      )

      val someSchema = List(
        StructField("prediction", IntegerType, true),
        StructField("label", DateType , true)
      )

      val someDF = spark.createDataFrame(
        spark.sparkContext.parallelize(someData),
        StructType(someSchema)
      )

    // Testing exception
     val caught = intercept[IllegalArgumentException] {
        throwError(spark,someDF)
      }

     assert(caught.getMessage.contains("Illegal arguments"))
   }
}

If I move throw new IllegalArgumentException(s"Illegal arguments") outside the map function call, the test passes.

How can the exception thrown by the 'throwError' function be tested?

apache-spark scala unit-testing exception
1 Answers

Catching exceptions at row level is not possible with sparkDF, If you use an RDD it is possible to achieve what you are trying to do.

Check this blog: https://www.nicolaferraro.me/2016/02/18/exception-handling-in-apache-spark/

Workaround for your problem:

def throwError(spark: SparkSession,df:DataFrame): Unit = {
        import spark.implicits._
        val countOfRowsBeforeCheck = df.count()
        val predictionAndLabels = df.select("prediction", "label").flatMap {
            case Row(prediction: Double, label: Double) => Iterator((prediction, label))
            case other => Iterator.empty
        }
        val countOfRowsAfterCheck = predictionAndLabels.count()
        if(countOfRowsAfterCheck != countOfRowsBeforeCheck){
            throw new IllegalArgumentException(s"Illegal arguments")
        }

        predictionAndLabels.show()
}

Hope this help !!

user_90823050
Related