definieren Eingangsparameter für UDF in Funken scala

Question

ich in scala arbeite und Funken

ich ein UDF bin definieren, hier ist es

def udfcrpentcd = udf((CORP_ENT_CD:String)=>{ 
    state_name match{ 
     case "IL1" if state_name.contains("IL1")=> "IL1" 
     case "OK1" if state_name.contains("OK1")=> "OK1" 
     case "TX1" if state_name.contains("TX1")=> "TX1" 
     case "NM1" if state_name.contains("NM1")=> "NM1" 
     case "MT1" if state_name.contains("MT1")=> "MT1" 
     case _ =>"Null" 
    }}) 




val local_masterdb =old_dataframe_temp_masterdbDataFrame.withColumn("new_columna_name_CORP_ENT_CD",udfcrpentcd(old_dataframe_temp_masterdbDataFrame("last_column_of_old_dataframe_DB_STATUS")+1)) 
    local_masterdb.show() 

jetzt, ich will die oben UDF wiederzuverwenden,

Ich möchte es universell machen, anstatt State_name zu vergleichen, muss ich eine Zeichenfolge übergeben und dann CRP_ENT_CD zurückgegeben ... das ist, was ich tun möchte.

ist dies der richtige Weg ....

def udfcrpentcd (input_parameter:String) = udf((CORP_ENT_CD:String)=>{ 
    input_parameter match{ 
     case "IL1" if input_parameter.contains("IL1")=> "IL1" 
     case "OK1" if input_parameter.contains("OK1")=> "OK1" 
     case "TX1" if input_parameter.contains("TX1")=> "TX1" 
     case "NM1" if input_parameter.contains("NM1")=> "NM1" 
     case "MT1" if input_parameter.contains("MT1")=> "MT1" 
     case _ =>"Null" 
    }}) 

wenn dies richtig dann, wie es zurückrufen? anyhelp über das Übergeben von Parametern

Answer 1

Hier ist ein Beispiel dafür, wie Sie Parameter an ein udf übergeben können.

val udfcrpentcd_res = udf(udfcrpentcd) 
def udfcrpentcd (String => String) = (input_parameter: String) =>{ 
input_parameter match{ 
    case "IL1" if input_parameter.contains("IL1")=> "IL1" 
    case "OK1" if input_parameter.contains("OK1")=> "OK1" 
    case "TX1" if input_parameter.contains("TX1")=> "TX1" 
    case "NM1" if input_parameter.contains("NM1")=> "NM1" 
    case "MT1" if input_parameter.contains("MT1")=> "MT1" 
    case _ =>"Null" 
}}) 

val local_masterdb = old_dataframe_temp_masterdbDataFrame.withColumn("new_columna_name_CORP_ENT_CD",udfcrpentcd_res(old_dataframe_temp_masterdbDataFrame("last_column_of_old_dataframe_DB_STATUS")+1)) 
local_masterdb.show()