通过 udf 进行拼写校正

Question

我需要使用 spark 更正一些拼写。不幸的是，像

这样天真的方法

val misspellings3 = misspellings1
    .withColumn("A", when('A === "error1", "replacement1").otherwise('A))
    .withColumn("A", when('A === "error1", "replacement1").otherwise('A))
    .withColumn("B", when(('B === "conditionC") and ('D === condition3), "replacementC").otherwise('B))

不适用于 spark

简单的情况（前两个例子）可以通过

很好地处理

val spellingMistakes = Map(
    "error1" -> "fix1"
  )

  val spellingNameCorrection: (String => String) = (t: String) => {
    titles.get(t) match {
      case Some(tt) => tt // correct spelling
      case None => t // keep original
    }
  }
  val spellingUDF = udf(spellingNameCorrection)

  val misspellings1 = hiddenSeasonalities
    .withColumn("A", spellingUDF('A))

但我不确定如何以良好且可概括的方式处理 UDF 中更复杂/链接的条件替换。如果它只是一个相当小的拼写列表 < 50，你会建议在 UDF 中对它们进行硬编码吗？

Answer 1

您可以使 UDF 接收多列：

val spellingCorrection2= udf((x: String, y: String) => if (x=="conditionC" && y=="conditionD") "replacementC" else x)
val misspellings3 = misspellings1.withColumn("B", spellingCorrection2($"B", $"C")

为了使其更加通用，您可以使用从两个条件的元组到字符串的映射，与第一种情况相同。

如果您想进一步概括它，则可以使用数据集映射。基本上创建一个包含相关列的案例 class，然后使用 as 将数据框转换为案例 class 的数据集。然后使用数据集映射，并在其中对输入数据使用模式匹配来生成相关校正并转换回数据帧。这应该更容易编写，但会产生性能成本。

Answer 2

现在我将使用以下似乎工作得很好并且更容易理解的方法：https://gist.github.com/rchukh/84ac39310b384abedb89c299b24b9306

Answer 3

如果 spellingMap 是包含正确拼写的地图，df 是数据框。

val df: DataFrame = _ 
val spellingMap = Map.empty[String, String] //fill it up yourself
val columnsWithSpellingMistakes = List("abc", "def")

像这样写一个UDF

def spellingCorrectionUDF(spellingMap:Map[String, String]) =
udf[(String), Row]((value: Row) =>
{
  val cellValue = value.getString(0)
  if(spellingMap.contains(cellValue)) spellingMap(cellValue)
  else cellValue
})

最后，您可以将它们称为

val newColumns = df.columns.map{
case columnName =>
  if(columnsWithSpellingMistakes.contains(columnName)) spellingCorrectionUDF(spellingMap)(Column(columnName)).as(columnName)
  else Column(columnName)
}
df.select(newColumns:_*)

通过 udf 进行拼写校正

spark spelling correction via udf

user-defined-functions

apache-spark

apache-spark-sql

udf

spark-dataframe