Обновите столбец данных, сравнив его с существующими данными в другом столбце, используя алгоритм Левенштейна

Question

Как я могу обновить столбец m_name с помощью алгоритма Левенштейна, чтобы заменить нули?

+--------------------+--------------------+-------------------+
|       original_name|              m_name|            created|
+--------------------+--------------------+-------------------+
|            New York|            New York|2017-08-01 09:33:40|
|            new york|                null|2017-08-01 15:15:06|
|       New York city|                null|2017-08-01 15:15:06|
|          california|          California|2017-09-01 09:33:40|
| California,000IU...|                null|2017-09-01 01:40:00|
|         Californiya|          California|2017-09-01 11:38:21|

Для каждого значения «original_name» должно быть взято первое ближайшее значение «m_name», основанное алгоритмом на основе расстояния Левенштейна (расстояние редактирования).

similarity(s1,s2) = [max(len(s1), len(s2)) − editDistance(s1,s2)] / max(len(s1), len(s2))

"идеальный" конечный результат должен быть таким:

+--------------------+--------------------+-------------------+
|       original_name|              m_name|            created|
+--------------------+--------------------+-------------------+
|            New York|            New York|2017-08-01 09:33:40|
|            new york|            New York|2017-08-01 15:15:06|
|       New York city|            New York|2017-08-01 15:15:06|
|          california|          California|2017-09-01 09:33:40|
| California,000IU...|          California|2017-09-01 01:40:00|
|         Californiya|          California|2017-09-01 11:38:21|

Answer 1

Кредит идет на rossettacode Levenshtein_distance

Вы можете сделать следующее (прокомментировано для ясности и объяснения)

//collecting the m_name to unique set and filtering out nulls and finally broadcasting to be used in udf function
import org.apache.spark.sql.functions._
val collectedList = df.select(collect_set("m_name")).rdd.collect().flatMap(row => row.getAs[Seq[String]](0).filterNot(_ == "null")).toList
val broadcastedList = sc.broadcast(collectedList)

//levenshtein distance formula applying
import scala.math.{min => mathmin, max => mathmax}
def minimum(i1: Int, i2: Int, i3: Int) = mathmin(mathmin(i1, i2), i3)

def editDistance(s1: String, s2: String) = {
  val dist = Array.tabulate(s2.length + 1, s1.length + 1) { (j, i) => if (j == 0) i else if (i == 0) j else 0 }

  for (j <- 1 to s2.length; i <- 1 to s1.length)
    dist(j)(i) = if (s2(j - 1) == s1(i - 1)) dist(j - 1)(i - 1)
    else minimum(dist(j - 1)(i) + 1, dist(j)(i - 1) + 1, dist(j - 1)(i - 1) + 1)

  dist(s2.length)(s1.length)
}

//udf function definition to find the levenshtein distance and finding the closest first match from the broadcasted list with original_name column
def levenshteinUdf = udf((str1: String)=> {
  val distances = for(str2 <- broadcastedList.value) yield (str2, editDistance(str1.toLowerCase, str2.toLowerCase))
  distances.minBy(_._2)._1
})


//calling the udf function when m_name is null
df.withColumn("m_name", when(col("m_name").isNull || col("m_name") === "null", levenshteinUdf(col("original_name"))).otherwise(col("m_name"))).show(false)

, что должно дать вам

+-------------------+----------+-------------------+
|original_name      |m_name    |created            |
+-------------------+----------+-------------------+
|New York           |New York  |2017-08-01 09:33:40|
|new york           |New York  |2017-08-01 15:15:06|
|New York city      |New York  |2017-08-01 15:15:06|
|california         |California|2017-09-01 09:33:40|
|California,000IU...|California|2017-09-01 01:40:00|
|Californiya        |California|2017-09-01 11:38:21|
+-------------------+----------+-------------------+

Примечание. Я не использовал вашу логику similarity(s1,s2) = [max(len(s1), len(s2)) − editDistance(s1,s2)] / max(len(s1), len(s2)), поскольку она дает неверный вывод