spark scala pattern matching on a dataframe column

user_2232892 0
Viewed 3594

I am coming from R background. I could able to implement the pattern search on a Dataframe col in R. But now struggling to do it in spark scala. Any help would be appreciated

problem statement is broken down into details just to describe it appropriately DF :

           Case                      Freq
            135322                     265
     183201,135322                      36
     135322,135322                      18
     135322,121200                      11
     121200,135322                       8
     112107,112107                       7
     183201,135322,135322                4
     112107,135322,183201,121200,80000   2

I am looking for a pattern search UDF, which gives me back all the matches of the pattern and then corresponding Freq value from the second col.

example : for pattern 135322 , i would like to find out all the matches in first col Case.It should return corresponding Freq number from Freq col. Like 265,36,18,11,8,4,2

for pattern 112107,112107 it should return just 7 because there is one matching pattern.

This is how the end result should look 

          Case                           Freq   results
            135322                       265    256+36+18+11+8+4+2
     183201,135322                        36    36+4+2
     135322,135322                        18    18+4
     135322,121200                        11    11+2
     121200,135322                         8    8+2
     112107,112107                         7    7
     183201,135322,135322                  4    4
     112107,135322,183201,121200,80000     2    2

what i tried so far:

val text= DF.select("case").collect().map(_.getString(0)).mkString("|")

 //search function for pattern search

 val valsum = udf((txt: String, pattern : String)=> { 
    txt.split("\\|").count(_.contains(pattern)) 
  } )

 //apply the UDF on the first col 
 val dfValSum = DF.withColumn("results", valsum( lit(text),DF("case")))
apache-spark scala regex
1 Answers

This one works

import common.Spark.sparkSession
import java.util.regex.Pattern
import util.control.Breaks._

object playground extends App {

  import org.apache.spark.sql.functions._

  val pattern = "135322,121200" // Pattern you want to search for

  // udf declaration
  val coder: ((String, String) => Boolean) = (caseCol: String, pattern: String) =>
    {
      var result = true
      val splitPattern = pattern.split(",")
      val splitCaseCol = caseCol.split(",")
      var foundAtIndex = -1

      for (i <- 0 to splitPattern.length - 1) {
        breakable {
          for (j <- 0 to splitCaseCol.length - 1) {
            if (j > foundAtIndex) {
              println(splitCaseCol(j))
              if (splitCaseCol(j) == splitPattern(i)) {
                result = true
                foundAtIndex = j
                break
              } else result = false
            } else result = false
          }
        }
      }
      println(caseCol, result)
      (result)
    }

  // registering the udf  
  val udfFilter = udf(coder)

  //reading the input file
  val df = sparkSession.read.option("delimiter", "\t").option("header", "true").csv("output.txt")

  //calling the function and aggregating
  df.filter(udfFilter(col("Case"), lit(pattern))).agg(lit(pattern), sum("Freq")).toDF("pattern","sum").show

}

if input is

135322,121200

Output is 

+-------------+----+
|      pattern| sum|
+-------------+----+
|135322,121200|13.0|
+-------------+----+

if input is

135322,135322

Output is 

+-------------+----+
|      pattern| sum|
+-------------+----+
|135322,135322|22.0|
+-------------+----+
user_45767950
Related