有效地加入和不加入数据框的数据与其他数据框

Question

我有两个数据框，比如说 A 和 B。它们有不同的模式。

我想从数据框 A 中获取记录，该数据框与 B 在一个键上连接，以及未连接的记录，我也想要这些。

这可以在单个查询中完成吗？因为两次遍历相同的数据会降低性能。 DataFrame A 的大小比 B 大得多。 Dataframe B 的大小约为 50Gb-100gb。因此在那种情况下我不能广播 B。

我可以接受单个 Dataframe C 作为结果，它可以有一个分区列 "Joined"，其值为 "Yes" 或 "No"，表示 A 中的数据是否得到是否加入B.

万一A有重复怎么办？我不想要他们。我在想稍后我会在 C 数据帧上做一个 recudeByKey。有什么建议吗？

我正在使用配置单元表将数据以 ORC 文件格式存储在 HDFS 上。在 scala 中编写代码。

Answer 1

是的，您只需要进行左外连接：

import sqlContext.implicits._

val A = sc.parallelize(List(("id1", 1234),("id1", 1234),("id3", 5678))).toDF("id1", "number")
val B = sc.parallelize(List(("id1", "Hello"),("id2", "world"))).toDF("id2", "text")

val joined = udf((id: String) => id match {
  case null => "No"
  case _ => "Yes"
})

val C = A
  .distinct
  .join(B, 'id1 === 'id2, "left_outer")
  .withColumn("joined",joined('id2))
  .drop('id2)
  .drop('text)

这将生成一个数据框 C:[id1: string, number: int, joined: string]，如下所示：

[id1,1234,Yes]
[id3,5678,No]

请注意，我添加了 distinct 以过滤掉 A 中的重复项，并且 C 中的最后一列表示是否已加入。

EDIT: Following remark from OP, I have added the drop lines to remove the columns from B.

有效地加入和不加入数据框的数据与其他数据框

efficiently get joined and not joined data of a dataframe against other dataframe

join

apache-spark

rdd

apache-spark-sql

spark-dataframe