apache-spark - 如何对两个模式 RDD 执行连接操作？

Question

我有两个 SchemaRDD，我想对它们执行连接操作（与 SQL 连接相同）。请帮我。

Answer 1

如果将两个 SchemaRDD 注册为表，则实际上可以进行 SQL 连接。虽然以下示例使用案例类，但该技术并不依赖于它们：

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext._
case class Score(name: String, score: Int)
case class Age(name: String, age: Int)
val scores = sc.textFile("scores.txt").map(_.split(",")).map(s => Score(s(0),s(1).trim.toInt))
val ages = sc.textFile("ages.txt").map(_.split(",")).map(s => Age(s(0),s(1).trim.toInt))
scores.registerAsTable("scores")
ages.registerAsTable("ages")
val joined = sqlContext.sql("""
    SELECT a.name, a.age, s.score
    FROM ages a JOIN scores s
    ON a.name = s.name""")
joined.collect().foreach(println)

另一种不需要注册的方法是使用语言集成语法（有人怀疑有一种方法可以简化这一点）：

val scoresAliased = scores.as('s)
val agesAliased = ages.as('a)
val joined = 
  scoresAliased.join(agesAliased, Inner, Some("s.name".attr === "a.name".attr))

Answer 2

从火花文档

join(otherDataset, [numTasks]) 当在 (K, V) 和 (K, W) 类型的数据集上调用时，返回 (K, (V, W)) 对的数据集，其中每个键的所有元素对。通过 leftOuterJoin 和 rightOuterJoin 也支持外连接。

您可能希望通过映射转换运行您的 schemaRDD，以将它们放入(K,V)表单中：K是您要加入的键，并且V可以只是整个行对象。例如

val a = ...
val b = ...
val bWithKey = b.map(v => (v.key, v))
val joined = a.map(v => (v.key, v)).join(bWithKey)