複数の列でデータセットを結合するにはどうすればよいですか？

Question

考えると2スパークデータセット、AとB、私は次のように単一の列に参加することが可能です。

a.joinWith(b, $"a.col" === $"b.col", "left") 

私の質問は、あなたが複数の列を使用して参加出来るかどうかです。基本的に以下のデータフレーム APIコードの同等：

a.join(b, a("col") === b("col") && a("col2") === b("col2"), "left") 

Answer 1

あなたは正確にDataframeと同じようにそれを行うことができます：スパーク<で

val xs = Seq(("a", "foo", 2.0), ("x", "bar", -1.0)).toDS 
val ys = Seq(("a", "foo", 2.0), ("y", "bar", 1.0)).toDS 

xs.joinWith(ys, xs("_1") === ys("_1") && xs("_2") === ys("_2"), "left").show 
// +------------+-----------+ 
// |   _1|   _2| 
// +------------+-----------+ 
// | [a,foo,2.0]|[a,foo,2.0]| 
// |[x,bar,-1.0]|  null| 
// +------------+-----------+ 

は、あなたがこのようなものを使用することができます2.0.0 ：

xs.as("xs").joinWith(
    ys.as("ys"), ($"xs._1" === $"ys._1") && ($"xs._2" === $"ys._2"), "left") 

Answer 2

whereを次々と連鎖させる別の方法があります。最初すなわち

scala> case class A(id: Long, name: String) 
defined class A 

scala> case class B(id: Long, name: String) 
defined class B 

scala> val as = Seq(A(0, "zero"), A(1, "one")).toDS 
as: org.apache.spark.sql.Dataset[A] = [id: bigint, name: string] 

scala> val bs = Seq(B(0, "zero"), B(1, "jeden")).toDS 
bs: org.apache.spark.sql.Dataset[B] = [id: bigint, name: string] 

scala> as.join(bs).where(as("id") === bs("id")).show 
+---+----+---+-----+ 
| id|name| id| name| 
+---+----+---+-----+ 
| 0|zero| 0| zero| 
| 1| one| 1|jeden| 
+---+----+---+-----+ 


scala> as.join(bs).where(as("id") === bs("id")).where(as("name") === bs("name")).show 
+---+----+---+----+ 
| id|name| id|name| 
+---+----+---+----+ 
| 0|zero| 0|zero| 
+---+----+---+----+ 

なグッディ理由はスパークオプティマイザが一つに（しゃれが意図していない）の連続where Sに参加することで、whereオペレータ（S）に続いて参加する（必要に応じてその種類）を指定しますjoinである。基礎となる論理計画と物理計画を確認するには、explain演算子を使用します。

scala> as.join(bs).where(as("id") === bs("id")).where(as("name") === bs("name")).explain(extended = true) 
== Parsed Logical Plan == 
Filter (name#31 = name#36) 
+- Filter (id#30L = id#35L) 
    +- Join Inner 
     :- LocalRelation [id#30L, name#31] 
     +- LocalRelation [id#35L, name#36] 

== Analyzed Logical Plan == 
id: bigint, name: string, id: bigint, name: string 
Filter (name#31 = name#36) 
+- Filter (id#30L = id#35L) 
    +- Join Inner 
     :- LocalRelation [id#30L, name#31] 
     +- LocalRelation [id#35L, name#36] 

== Optimized Logical Plan == 
Join Inner, ((name#31 = name#36) && (id#30L = id#35L)) 
:- Filter isnotnull(name#31) 
: +- LocalRelation [id#30L, name#31] 
+- Filter isnotnull(name#36) 
    +- LocalRelation [id#35L, name#36] 

== Physical Plan == 
*BroadcastHashJoin [name#31, id#30L], [name#36, id#35L], Inner, BuildRight 
:- *Filter isnotnull(name#31) 
: +- LocalTableScan [id#30L, name#31] 
+- BroadcastExchange HashedRelationBroadcastMode(List(input[1, string, false], input[0, bigint, false])) 
    +- *Filter isnotnull(name#36) 
     +- LocalTableScan [id#35L, name#36]