在 Spark 中加入两个数据帧正在爆炸

Question

我正在尝试在一个字段上加入两个数据框。为此，我必须首先确保该字段是唯一的。所以我的事件顺序是：

1. 读入第一个数据帧
2. 选择我要加入的字段（例如，field1），以及我要加入的另一个字段（field2）
3. 做.distinct

那么，对于第二张桌子..

1. 读入第二个数据帧
2. 在field1 上与第一个表进行左连接
3. 做.distinct

我尝试运行我的脚本，但它花费的时间比它应该的要长。 为了调试它，我在连接前后的第一个表上添加了println 记录计数，结果如下：

在加入之前，记录数为904,326。之后，2,658,632。 所以我认为它正在爆炸，但不知道为什么。我认为这与选择两个字段后尝试仅使用一个“不同”有关..？

请帮忙！

代码如下：

    val ticketProduct = Source.fromArg(args, "f1").read
   .select($"INSTRUMENT_SK", $"TICKET_CODES_SK")
   .distinct

  val instrumentD = Source.fromArg(args, "f2").read
//    println("instrumentD count before join is " + instrumentD.count)
   .join(ticketProduct, Seq("INSTRUMENT_SK"), "leftouter")
//   .select($"SERIAL_NBR", $"TICKET_CODES_SK")
   .distinct        
    println("instrumentD count after join is " + instrumentD.count)

Answer 1

您遇到的问题是，通过调用 distinct 您只能删除 field1 和 field2 的值相同的行。 由于您加入 field1，您可能希望 field1 的值是唯一的。

您可以尝试以下方式，而不是调用distinct。

dataframe1.groupBy($"field1").agg(org.apache.spark.sql.functions.array($"field2"))

这将产生一个数据框，其中 field1 列是唯一的，field2 的多个值聚合到一个数组中。

这同样适用于第二个数据帧。

举个例子：假设您有包含以下内容的数据框。

field1, field2
 1,       1
 1,       2

field1, field3
 1,      1
 1,      3

然后 distinct 对它们没有任何作用，因为行是不同的。 现在，如果您要加入 field1，您将获得以下信息。

 1, 1, 1
 1, 2, 1
 1, 1, 3
 1, 2, 3

相反的聚合会给出

field1, array_of_field2
 1,      [1,2]

field1, array_of field3
 1,      [1,3]

然后连接将产生以下数据帧。

1, [1,2], [1,3]