Is there a way to pass default case value as an argument

user_6388994 0
Viewed 147

I'm trying to see if there is a way that I can incorporate the default case's value in a match expression

item match {
case null => false
case _:String => _.matches(params)
}

I've tried looking at different options, but I couldn't really find a solution.

For context this is part of this snippet of code, performing it on a spark dataframe

x.filter(row => row.getAs[String]("colname").matches("""/regex/""")).count()

but there are ocassional null values that make everything break.

I know I can do this to get the desired result, but doesn't make sense to calculate it again

x.filter(row => row.getAs[String]("colname") match{
         case null => false
         case _ => row.getAs[String]("colname").matches("""\d""")
         }).count()

Any suggestions or pointers would be helpful. Thank you

apache-spark scala
4 Answers

To match everything you just need to use variable name without type ascription:

item match {
   case null => false
   case i => i.matches(params)
}

So in your case:

x.filter(row => 
   row.getAs[String]("colname") match {
         case null => false
         case r => r.matches("""\d""")
   }
).count()

If you want to check if the item is null, you could also use Option.apply which returns None if the value is null and Some otherwise, for example:

Option(item).fold(false)(_.matches(params))
user_13033240

In addition to Krzysztof Atłasik answer, you can add the check for null before the match-statament, you could do:

x
.where($"colname".isNotNull)
.filter(row => row.getAs[String]("colname").matches("""/regex/""")).count()

or something like

x.where($"colname" rlike """\d""").count()
user_11385230

You can just use rlike. For example:

x.filter(x("colName").rlike("cat|dog")).count
user_31299770

If I understood your requirement correctly, you are looking to filter and aggregate in a single iteration and I think UDAF is something you will have to go for. Essentially, I don't see any performance benefits of doing this over the option mentioned at the end of your original post i.e. udf to filter out and then a count. However, I have demonstrated how the UDAF can be used here.

For Spark<3.0, you will have to extend the UserDefinedAggregateFunction class to look somewhat like this -

    import org.apache.spark.sql.Row
    import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
    import org.apache.spark.sql.types._

    class ConditionalCount(val pattern: String, colName: String) extends UserDefinedAggregateFunction{
      override def inputSchema: StructType = StructType(Seq(StructField(colName, StringType)))

      override def bufferSchema: StructType = StructType(Seq(StructField("count", LongType)))

      override def dataType: DataType = LongType

      override def deterministic: Boolean = true

      override def initialize(buffer: MutableAggregationBuffer): Unit = buffer(0) = 0L

      override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
        buffer(0) = buffer.getAs[Long](0) + input.getAs[String](0) match {
          case null => 0
          case x if x.matches(pattern) => 1L
          case _ => 0L
        }
      }

      override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit =
        buffer1(0) = buffer1.getAs[Long](0) + buffer2.getAs[Long](0)

      override def evaluate(buffer: Row): Any = buffer.getAs[Long](0)
    }

Then use this initialize and use this UDAF to look like something like this -


    val conditionalCount = new ConditionalCount("my pattern", "colName")

    df.withColumn("count", conditionalCount(col("colName")))

P.S.: I have not run this and verified, pardon me if there are minute errors

user_19456360
Related