我是 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 类型更改为 Map[(String,String,String),String]
时,我不知道您是如何在函数中传递它的。
万分感谢。
最佳答案
如果您尝试为 A1
与 B2
和 A2
与 的每次匹配获取
和 B8
值code>B6A3
与 B9
,然后简单的 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 |
+---+
希望回答对你有帮助
已更新
根据我从您的问题和下面的评论中了解到的情况,您对如何将 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 |
+----------+------+------+-----------+
关于sql - 如何在spark中使用多个键构建查找功能,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/45150093/