如何在火花中使用多个键构建查找功能

Question

我是 spark 的新手，上周问了一个类似的问题。它编译但不工作。所以我真的不知道该怎么办。这是我的问题：我的表 A 包含 3 列，如下所示

-----------
A1  A1  A3
-----------
a    b   c

和另一个像这样的表 B

------------------------------------
B1  B2  B3  B4  B5  B6  B7  B8  B9
------------------------------------
1   a   3   4   5   b   7   8    c

我的逻辑是：A1 A2 A3 是我的键，它对应于表 B 中的 B2 B6 B9。我需要构建一个以 A1 A2 A3 为键并返回 B8 的查找函数。

这是我上周尝试的：

//getting the data in to dataframe
val clsrowRDD = clsfile.map(_.split("\t")).map(p => Row(p(0),p(1),p(2),p(3),p(4),p(5),p(6),p(7),p(8)))
val clsDataFrame = sqlContext.createDataFrame(clsrowRDD, clsschema)

//mapping the three key with the value
val smallRdd = clsDataFrame.rdd.map{row: Row => (mutable.WrappedArray.make[String](Array(row.getString(1), row.getString(5), row.getString(8))), row.getString(7))}

val lookupMap:Map[mutable.WrappedArray[String], String] = smallRdd.collectAsMap()

//build the look up function
def lookup(lookupMap: Map[mutable.WrappedArray[String],String]) =
udf((input: mutable.WrappedArray[String]) => lookupMap.lift(input))

//call the function
val combinedDF  = mstrDataFrame.withColumn("ENTP_CLS_CD",lookup(lookupMap)($"SRC_SYS_CD",$"ORG_ID",$"ORG_CD"))

这段代码可以编译，但并没有真正返回我需要的结果。我在想这是因为我传入了一个数组作为键，而我的表中并没有真正的数组。但是当我尝试将地图类型更改为Map[(String,String,String),String]时，我不知道你是如何在函数中传递它的。

非常感谢。

Answer 1

如果您尝试为A1 与B2 和A2 与B6 和A3 与B9 的每个匹配获取B8 值，那么简单的join 和select方法应该可以解决问题。 创建查找地图会产生复杂性。

正如您所解释的，您必须将数据帧 df1 和 df2 作为

+---+---+---+
|A1 |A2 |A3 |
+---+---+---+
|a  |b  |c  |
+---+---+---+

+---+---+---+---+---+---+---+---+---+
|B1 |B2 |B3 |B4 |B5 |B6 |B7 |B8 |B9 |
+---+---+---+---+---+---+---+---+---+
|1  |a  |3  |4  |5  |b  |7  |8  |c  |
|1  |a  |3  |4  |5  |b  |7  |8  |e  |
+---+---+---+---+---+---+---+---+---+

简单的join和select都可以搞定

df1.join(df2, $"A1" === $"B2" && $"A2" === $"B6" && $"A3" === $"B9", "inner").select("B8")

这应该给你

+---+
|B8 |
+---+
|8  |
+---+

希望回答对你有帮助

更新

根据我从您的问题和下面的 cmets 中了解到的情况，您对如何将 array 传递给您的 lookup udf 函数感到困惑。为此，您可以使用 array 函数。我已经修改了您几乎完美的代码的某些部分以使其正常工作

//mapping the three key with the value
val smallRdd = clsDataFrame.rdd
  .map{row: Row => (mutable.WrappedArray.make[String](Array(row.getString(1), row.getString(5), row.getString(8))), row.getString(7))}

val lookupMap: collection.Map[mutable.WrappedArray[String], String] = smallRdd.collectAsMap()

//build the look up function
def lookup(lookupMap: collection.Map[mutable.WrappedArray[String],String]) =
udf((input: mutable.WrappedArray[String]) => lookupMap.lift(input))

//call the function
val combinedDF  = mstrDataFrame.withColumn("ENTP_CLS_CD",lookup(lookupMap)(array($"SRC_SYS_CD",$"ORG_ID",$"ORG_CD")))

你应该有

+----------+------+------+-----------+
|SRC_SYS_CD|ORG_ID|ORG_CD|ENTP_CLS_CD|
+----------+------+------+-----------+
|a         |b     |c     |8          |
+----------+------+------+-----------+