Соответствие регулярному выражению со значениями столбца данных

Question

Я хочу выполнить поиск между Map[String,List[scala.util.matching.Regex]] со столбцом данных. если какое-либо из List[scala.util.matching.Regex] совпадает со значениями столбца dataframe, то должно возвращаться key из Map[String,List[scala.util.matching.Regex]]

Map[String,List[scala.util.matching.Regex]] = Map(m1 -> List(rule1, rule2), m2 -> List(rule3), m3 -> List(rule6)))

Я хочу перебрать список регулярных выражений и сопоставить со значением столбца dataframe , было бы лучше, если бы сравнение с регулярным выражением можно было выполнять параллельно, а не последовательно

dataframe


+------------------------+
|desc                    |
+------------------------+
|STRING MATCHES SSS rule1|
|STRING MATCHES SSS rule1|
|STRING MATCHES SSS rule1|
|STRING MATCHES SSS rule2|
|STRING MATCHES SSS rule2|
|STRING MATCHES SSS rule3|
|STRING MATCHES SSS rule3|
|STRING MATCHES SSS rule6|
+------------------------+

O / P:

+-------------------+------------------------+
|merchant           |desc                    |
+-------------------+------------------------+
|m1                 |STRING MATCHES SSS rule1|
|m1                 |STRING MATCHES SSS rule1|
|m1                 |STRING MATCHES SSS rule1|
|m1                 |STRING MATCHES SSS rule2|
|m1                 |STRING MATCHES SSS rule2|
|m2                 |STRING MATCHES SSS rule3|
|m2                 |STRING MATCHES SSS rule3|
|m3                 |STRING MATCHES SSS rule6|
+-------------------+------------------------+

Answer 1

Вот еще один способ, основанный на функции DataFrame map и предопределенном наборе правил rules:

import spark.implicits._
import scala.util.matching.Regex

val df = Seq(
("STRING MATCHES SSS rule1"),
("STRING MATCHES SSS rule1"),
("STRING MATCHES SSS rule1"),
("STRING MATCHES SSS rule2"),
("STRING MATCHES SSS rule2"),
("STRING MATCHES SSS rule3"),
("STRING MATCHES SSS rule3"),
("STRING MATCHES SSS rule6"),
("STRING MATCHES SSS ruleXXX")
).toDF("desc")

val rules = Map(
  "m1" -> List("rule1".r, "rule2".r), 
  "m2" -> List("rule3".r), 
  "m3" -> List("rule6".r)
)

df.map{r =>
  val desc = r.getString(0)
  val merchant = rules.find(_._2.exists(_.findFirstIn(desc).isDefined)) match {
      case Some((m : String, _)) => m
      case None => null
    }

  (merchant, desc)
}.toDF("merchant", "desc").show(false)

Вывод:

+--------+--------------------------+
|merchant|desc                      |
+--------+--------------------------+
|m1      |STRING MATCHES SSS rule1  |
|m1      |STRING MATCHES SSS rule1  |
|m1      |STRING MATCHES SSS rule1  |
|m1      |STRING MATCHES SSS rule2  |
|m1      |STRING MATCHES SSS rule2  |
|m2      |STRING MATCHES SSS rule3  |
|m2      |STRING MATCHES SSS rule3  |
|m3      |STRING MATCHES SSS rule6  |
|null    |STRING MATCHES SSS ruleXXX|
+--------+--------------------------+

Объяснение:

• rules.find(... из правил найдите пару ключ / значение

• _._2.exists(..., значение которой имеет регулярное выражение

• _.findFirstIn(desc).isDefined, совпадающее с desc

• case Some((m : String, _)) => m и извлечение ключа из этой пары

PS: я не уверен, что Вы имеете в виду совпадение с регулярным выражением может быть выполнено параллельно, а не последовательно , так как функция отображения в приведенном выше решении будет выполняться параллельно. Уровень распараллеливания зависит от выбранного номера раздела. Добавление дополнительного распараллеливания внутри функции карты, например, в виде потока (или Scala Futures), безусловно, усложнит код без увеличения производительности. Это потому, что если вы создаете большое количество потоков, более вероятно, что вы создадите узкое место для ЦП вместо ускорения вашей программы. Spark - эффективная распределенная система, и нет необходимости искать альтернативы в отношении параллельного выполнения.

Answer 2

Вы можете объявить UDF, как показано ниже, который будет работать параллельно и будет быстрым. Ниже приведена только ссылка, как я понял вашу проблему. Вы можете взять это в качестве справки и соответственно оформить свой UDF.

scala> import org.apache.spark.sql.expressions.UserDefinedFunction

scala> def RuleCheck:UserDefinedFunction = udf((colmn:String) => {
     |  val Rule:Map[String,List[String]] = Map("Number" -> List("[0-9]"),"Statment" -> List("[a-zA-Z]"), "Fruit" -> List("apple","banana","orange"), "Country" -> List("India","US","UK"))
     | var Out = scala.collection.mutable.Set[String]()
     | Rule.foreach{ rr =>
     | val key = rr._1
     | val Listrgx = rr._2
     | 
     | Listrgx.foreach{ x =>
     | val rgx = x.r
     | 
     | if(rgx.findFirstMatchIn(colmn).mkString != ""){
     | Out += key
     | }
     |         }
     |       }
     |       Out.mkString(",") })

scala> df.show()
+---+--------------------+
| id|             comment|
+---+--------------------+
|  1|     I have 3 apples|
|  2|I like banana and...|
|  3|        I am from US|
|  4|          1932409243|
|  5|       I like orange|
|  6|         #%@#$@#%@#$|
+---+--------------------+


scala> df.withColumn("Key", RuleCheck(col("comment"))).show(false)
+---+---------------------------------+----------------------+
|id |comment                          |Key                   |
+---+---------------------------------+----------------------+
|1  |I have 3 apples                  |Number,Fruit,Statment |
|2  |I like banana and I am from India|Country,Fruit,Statment|
|3  |I am from US                     |Country,Statment      |
|4  |1932409243                       |Number                |
|5  |I like orange                    |Fruit,Statment        |
|6  |#%@#$@#%@#$                      |                      |
+---+---------------------------------+----------------------+