如何执行“查找”关于星火dataframes给出多个条件
问题描述:
我在星火新手操作(我的版本是1.6.0),现在我试图解决如下问题:如何执行“查找”关于星火dataframes给出多个条件
假设有是两个源文件:
- 第一个(简称A)是一个大的包含名为A1,B1,C1和其他80列的列。里面有23万条记录。
- 第二个(简称B)是一个小查找表,其中包含名为A2,B2,C2和D2的列。里面有250条记录。
现在我们需要插入一个新的柱分为A,下面给出的逻辑:
- 第一查找A1,B1和(对应列是A2,B2和C2)C1在B,如果成功的,返回D2作为新添加的列的值。如果没有发现...
- 然后在B中查找A1,B1。如果成功,则返回D2。如果没有发现任何...
- 设置默认值“NA”
我已经阅读文件和它们改建为数据帧。对于第一种情况,我通过将左边的外部结合在一起得到了结果。但我在下一步找不到好的方法。
我目前的尝试是通过使用不太严格的条件连接A和B来构建新的数据帧。但我不知道如何从另一个更新当前数据帧。或者还有其他更直观,更有效的方法来解决整个问题吗?
感谢您的所有答案。
----------------------------- 20160309更新-------------- ------------------
最后接受@mlk的答案。还是非常感谢@ zero323对UDF和加入的评论,钨代码的生成实际上是我们现在面临的另一个问题。但是,因为我们需要为每一个查询做查询的分数和平均4个条件,前者的解决方案更适合...
最终的解决方案是某种看起来像下面片段:
```
import sqlContext.implicits._
import com.github.marklister.collections.io._
case class TableType(A: String, B: String, C: String, D: String)
val tableBroadcast = sparkContext.broadcast(CsvParser(TableType).parseFile("..."))
val lkupD = udf {
(aStr: String, bStr: String, cStr: String) =>
tableBroadcast.value.find {
case TableType(a, b, c, _) =>
(a == aStr && b == bStr && c == cStr) ||
(a == aStr && b == bStr)
}.getOrElse(TableType("", "", "", "NA")).D
}
df = df.withColumn("NEW_COL", lkupD($"A", $"B", $"C"))
```
答
由于B是小我认为做到这一点的最佳方式是广播变量和用户定义的功能。
// However you get the data...
case class BType(A2: Int, B2: Int, C2 : Int, D2 : String)
val B = Seq(BType(1,1,1,"B111"), BType(1,1,2, "B112"), BType(2,0,0, "B200"))
val A = sc.parallelize(Seq((1,1,1, "DATA"), (1,1,2, "DATA"), (2, 0, 0, "DATA"), (2, 0, 1, "NONE"), (3, 0, 0, "NONE"))).toDF("A1", "B1", "C1", "OTHER")
// Broadcast B so all nodes have a copy of it.
val Bbradcast = sc.broadcast(B)
// A user defined function to find the value for D2. This I'm sure could be improved by whacking it into maps. But this is a small example.
val findD = udf {(a: Int, b : Int, c: Int) => Bbradcast.value.find(x => x.A2 == a && x.B2 == b && x.C2 == c).getOrElse(Bbradcast.value.find(x => x.A2 == a && x.B2 == b).getOrElse(BType(0,0,0,"NA"))).D2 }
// Use the UDF in a select
A.select($"A1", $"B1", $"C1", $"OTHER", findD($"A1", $"B1", $"C1").as("D")).show
答
只是为了参考,而不UDF的一个解决方案:
val b1 = broadcast(b.toDF("A2_1", "B2_1", "C2_1", "D_1"))
val b2 = broadcast(b.toDF("A2_2", "B2_2", "C2_2", "D_2"))
// Match A, B and C
val expr1 = ($"A1" === $"A2_1") && ($"B1" === $"B2_1") && ($"C1" === $"C2_1")
// Match A and B mismatch C
val expr2 = ($"A1" === $"A2_2") && ($"B1" === $"B2_2") && ($"C1" !== $"C2_2")
val toDrop = b1.columns ++ b2.columns
toDrop.foldLeft(a
.join(b1, expr1, "leftouter")
.join(b2, expr2, "leftouter")
// If there is match on A, B, C then D_1 should be not NULL
// otherwise we fall-back to D_2
.withColumn("D", coalesce($"D_1", $"D_2"))
)((df, c) => df.drop(c))
这假设有在每个类别至多一个匹配(所有三列,或前两个)或重复的行中的输出是期望。
UDF VS JOIN:
有多种因素需要考虑,并没有简单的答案在这里:
缺点:
- 广播
joins
需要传递数据的两倍到工人节点。至于broadcasted
表格没有被缓存(SPARK-3863),并且不可能在最近的将来改变(Resolution:Later)。 -
join
即使有完全匹配,操作也会应用两次。
优点:
-
join
3210和是透明的优化而UDF是不。 - 直接使用SQL表达式操作可以受益于所有的Tungsten优化,包括代码生成,而UDF不能。
这可能是要走的路。我也用'连接'提供了一个替代解决方案。 – zero323
谢谢mlk。如果查找表很大(500K * 50),播放它还是不错的? –
我的另一个问题是,假设我需要在不同的列上进行30次查找,并编写50个UDF,性能是否会受到影响? –