在 Spark 中查找 table

Question

我在 Spark 中有一个数据框，没有明确定义的模式，我想将其用作查找 table。例如，下面的数据框：

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

我是说 "schema not clearly defined" 因为在读取数据时键的数量是未知的，所以我把它留给 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
})

但是它抛出空指针异常错误。

有人可以告诉我 UDF 有什么问题吗？如果有 better/simpler 在 Spark 中执行此查找的方法？

Answer 1

我假设查找 table 非常小，在这种情况下，将其收集到驱动程序并将其转换为正常 Map 会更有意义。然后在UDF函数中使用这个Map。它可以通过多种方式完成，例如像这样：

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

使用上面的 lookup_map 变量，UDF 将只是：

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

最终的dataframe可以通过以下方式获得：

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