如何连接两个数据框？

Question

我无法让 Sparks DataFrame join 工作（没有产生结果）。这是我的代码：

val e = Seq((1, 2), (1, 3), (2, 4))
var edges = e.map(p => Edge(p._1, p._2)).toDF()
var filtered = edges.filter("start = 1").distinct()
println("filtered")
filtered.show()
filtered.printSchema()
println("edges")
edges.show()
edges.printSchema()
var joined = filtered.join(edges, filtered("end") === edges("start"))//.select(filtered("start"), edges("end"))
println("joined")
joined.show()

它需要在顶层定义case class Edge(start: Int, end: Int)。这是它产生的输出：

filtered
+-----+---+
|start|end|
+-----+---+
|    1|  2|
|    1|  3|
+-----+---+

root
 |-- start: integer (nullable = false)
 |-- end: integer (nullable = false)

edges
+-----+---+
|start|end|
+-----+---+
|    1|  2|
|    1|  3|
|    2|  4|
+-----+---+

root
 |-- start: integer (nullable = false)
 |-- end: integer (nullable = false)

joined
+-----+---+-----+---+
|start|end|start|end|
+-----+---+-----+---+
+-----+---+-----+---+

我不明白为什么输出是空的。为什么filtered 的第一行没有与edges 的最后一行合并？

Answer 1

val f2 = filtered.withColumnRenamed("start", "fStart").withColumnRenamed("end", "fEnd")
f2.join(edges, f2("fEnd") === edges("start")).show

我相信这是因为filtered("start").equals(edges("start"))，即filtered 是边缘的过滤视图，它们共享列定义。这些列是相同的，因此 Spark 无法识别您引用的是哪一个。

因此，您可以执行以下操作

edges.select(filtered("start")).show