Sparkのルックアップテーブル

Question

私はルックアップテーブルとして使用する明確に定義されたスキーマがないSparkのデータフレームを持っています。例えば、以下のデータフレーム：

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

は、スキーマは次のようになります。

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

データがされている間、キーの数が不明であるため、私は、「スキーマが明確に定義されていない」と言っています私はSparkにスキーマを推論するためにそれを残します。今

、私は以下のように列を持つ別のデータフレームがある場合：

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

を、私は結果になりたい：

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

私はこのようなUDFを試してみました：

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

しかし、Null Pointer例外エラーが発生します。

誰かがUDFの何が問題なのかを教えてもらえますか？また、Sparkでこの検索を行うためのより簡単な方法があれば教えてください。

Answer 1

ルックアップテーブルが非常に小さいと仮定します。この場合、ドライバに収集して通常のMapに変換する方が意味があります。次に、UDF機能でこのMapを使用します。

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

、最終データフレームをすることによって得ることができる：上記lookup_map変数を使用して

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

、UDFは単純に次のようになります。

それはこのような例のために、多くの方法で行うことができます

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