Lookup-Tabelle in Spark

Question

Ich habe ein Datenframe in Spark mit keinem klar definierten Schema, das ich als Nachschlagetabelle verwenden möchte. Zum Beispiel kann der Datenrahmen unter:

+------------------------------------------------------------------------+ 
|lookupcolumn               | 
+------------------------------------------------------------------------+ 
|[val1,val2,val3,val4,val5,val6]           | 
+------------------------------------------------------------------------+ 

Das Schema würde wie folgt aussehen:

|-- lookupcolumn: struct (nullable = true) 
| |-- key1: string (nullable = true) 
| |-- key2: string (nullable = true) 
| |-- key3: string (nullable = true) 
| |-- key4: string (nullable = true) 
| |-- key5: string (nullable = true) 
| |-- key6: string (nullable = true) 

Ich sage „Schema nicht klar definiert“, da die Anzahl der Schlüssel nicht bekannt ist, während die Daten werden lese, also überlasse ich es Spark, das Schema zu folgern.

Nun, wenn ich einen anderen Datenrahmen mit einer Spalte, wie unten:

+-----------------+ 
|  datacolumn| 
+-----------------+ 
|   key1 | 
|   key3 | 
|   key5 | 
|   key2 | 
|   key4 | 
+-----------------+ 

und ich möchte das Ergebnis sein:

+-----------------+ 
|  resultcolumn| 
+-----------------+ 
|   val1 | 
|   val3 | 
|   val5 | 
|   val2 | 
|   val4 | 
+-----------------+ 

ich ein UDF wie folgt versucht:

val get_val = udf((keyindex: String) => { 
    val res = lookupDf.select($"lookupcolumn"(keyindex).alias("result")) 
    res.head.toString 
}) 

Aber es wirft einen Ausnahmefehler Nullzeiger.

Kann mir jemand sagen, was mit der los ist, und ob es eine bessere/einfachere Methode gibt, diese Suche in Spark durchzuführen?

Answer 1

Ich nehme an, dass die Nachschlagetabelle ziemlich klein ist, in diesem Fall wäre es sinnvoller, sie zum Treiber zu sammeln und in einen normalen Map zu konvertieren. Dann verwenden Sie diese Map in der Funktion. Es kann in vielen Art und Weise, beispielsweise wie folgt geschehen:

val values = lookupDf.select("lookupcolumn.*").head.toSeq.map(_.toString) 
val keys = lookupDf.select("lookupcolumn.*").columns 
val lookup_map = keys.zip(values).toMap 

die oben lookup_map Variable, die UDF einfach sein:

val lookup = udf((key: String) => lookup_map.get(key)) 

Und der letzte Datenrahmen ist erhältlich:

val df2 = df.withColumn("resultcolumn", lookup($"datacolumn"))