通过udf火花拼写校正

Question

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

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)) 

一个天真的做法不符合火花工作How to add new columns based on conditions (without facing JaninoRuntimeException or OutOfMemoryError)?

简单的案件（第2例），可以很好地通过

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

现在我会去，这似乎只是正常工作，并更容易理解如下：https://gist.github.com/rchukh/84ac39310b384abedb89c299b24b9306

Answer 2

可以使UDF收到多列：

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

为了使这个更广义的，你可以使用地图从两个条件的元组为一个字符串一样的，你做的第一例。

如果你想更一般化它，那么你可以使用数据集映射。基本上用相关的列创建一个case类，然后使用as将数据框转换为case类的数据集。然后使用数据集图，并在其中使用输入数据的模式匹配来生成相关更正并转换回数据框。 这应该更容易编写，但会有性能成本。

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:_*)