在 Spark 中重用连接的数据框

Question

我在本地运行 HDFS 和 Spark，并试图了解 Spark 持久性的工作原理。我的目标是将连接的数据集存储在内存中，然后动态运行查询。但是，我的查询似乎是重做连接，而不是简单地扫描持久的预连接数据集。

通过从 HDFS 加载两个 CSV 文件，我创建并保留了两个数据帧，比如说 df1 和 df2。我将两个数据帧的连接保留在内存中：

val result = df1.join(df2, "USERNAME")
result.persist()
result.count()

然后我在结果之上定义一些操作：

val result2 = result.select("FOO", "BAR").groupBy("FOO").sum("BAR")
result2.show()

'result2' 不会捎带持久化的结果，而是自行重做连接。以下是 result 和 result2 的物理计划：

== Physical Plan for result ==
InMemoryColumnarTableScan [...], (InMemoryRelation [...], true, 10000, StorageLevel(true, true, false, true, 1), (TungstenProject [...]), None)


== Physical Plan for result2 ==
TungstenAggregate(key=[FOO#2], functions=[(sum(cast(BAR#10 as double)),mode=Final,isDistinct=false)], output=[FOO#2,sum(BAR)#837])
 TungstenExchange hashpartitioning(FOO#2)
  TungstenAggregate(key=[FOO#2], functions=[(sum(cast(BAR#10 as double)),mode=Partial,isDistinct=false)], output=[FOO#2,currentSum#1311])
   InMemoryColumnarTableScan [FOO#2,BAR#10], (InMemoryRelation [...], true, 10000, StorageLevel(true, true, false, true, 1), (TungstenProject [...]), None)

我会天真地假设，由于连接已经完成并在内存中分区，第二个操作将简单地由每个分区上的聚合操作组成。从头开始重做连接应该更昂贵。我是在假设错误还是做错了什么？此外，这是保留连接数据集以供以后查询的正确模式吗？

编辑：作为记录，在我关闭随机分区的数量后，第二个查询的性能提高了很多。默认情况下，spark.sql.shuffle.partitions 设置为 200。只需在我的本地实例上将其设置为 1 即可显着提高性能。

Answer 1

如果我们查看计划，我们会发现 Spark 实际上正在使用缓存的数据，而不是重做连接。自下而上：

这是 Spark 从缓存中读取数据：

InMemoryColumnarTableScan [FOO#2,BAR#10], (InMemoryRelation ...

这是 Spark 在每个分区中按 FOO 聚合 BAR - 查找 mode=Partial

TungstenAggregate(key=[FOO#2], functions=[(sum(cast(BAR#10 as double)),mode=Partial ...

这是 Spark 对上一步每个分区的数据进行洗牌：

TungstenExchange hashpartitioning(FOO#2)

这是 Spark 聚合已洗牌的分区总和 - 查找 mode=Final

TungstenAggregate(key=[FOO#2], functions=[(sum(cast(BAR#10 as double)),mode=Final ...

阅读这些计划有点痛苦，所以如果您可以访问 Spark UI 的 SQL 选项卡（我认为是 1.5+），我建议您改用它。