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