Apache Spark 中的 lookup() 函数

Question

我有两个数据集，一个存储在 Hive 中（较小的一个进一步用作查找表），另一个来自 spark 流。

现在我的要求是对这两个数据集执行一些操作。

例如：

dataset1：（存储在hive中）

id     name  
101    steve
102    david

dataset2：（来自火花流）

id   deprt   address

101   E01    NewYork
102   E02    London

每当我从流媒体中获得101 E01 NewYork 时，我都想合并两个数据集并返回如下结果：

id  name  dept  address
101 steve E01  NewYork

之前，我使用 Broadcast 变量完成了此类任务，我正在尝试通过使用 Hive 查找表来探索这一点。

谁能建议我如何做到这一点？

提前致谢。

Answer 1

假设您的配置单元查找表很小，因为您已经将其用作广播变量，您可以通过读取配置单元表来创建数据帧，您可以使用它来执行查找。

val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)
val lookupDF = hiveContext.sql("select * from your_hive_table").cache() //cache lookup data


ds.transform {
    rdd => 
        val df = rdd.toDF("c1","c2","c3")
        df.join(lookupDF, lookupDF("col") === df("col")).select("co1","col2","col3").rdd
}