Spark中连接两个RDD

我有两个 rdd，一个 rdd 只有一列，其他有两列来连接键上的两个 RDD 我添加了虚拟值 0 ，是否有其他有效的方法可以使用 join 来执行此操作？

val lines = sc.textFile("ml-100k/u.data")
val movienamesfile = sc.textFile("Cml-100k/u.item")

val moviesid = lines.map(x => x.split("\t")).map(x => (x(1),0))
val test = moviesid.map(x => x._1)
val movienames = movienamesfile.map(x => x.split("\\|")).map(x => (x(0),x(1)))
val shit = movienames.join(moviesid).distinct()

Edit:

让我把这个问题转换成SQL。比如说我有table1 (moveid) and table2 (movieid,moviename)。在 SQL 中我们这样写：

select moviename, movieid, count(1)
from table2 inner join table table1 on table1.movieid=table2.moveid 
group by ....

在 SQL 中table1只有一列，其中table2有两列仍然是join工作原理，Spark 中的相同方式可以连接来自两个 RDD 的键。

连接操作仅定义在PairwiseRDDs这与 SQL 中的关系/表有很大不同。的每个元素PairwiseRDD is a Tuple2其中第一个元素是key第二个是value。两者都可以包含复杂的对象，只要key提供了有意义的hashCode

如果你想用 SQL 语言来思考这个问题，你可以将 key 视为所有的内容ON条款和value包含选定的列。

SELECT table1.value, table2.value
FROM table1 JOIN table2 ON table1.key = table2.key

虽然这些方法乍一看很相似，并且您可以使用另一种方法来表达，但它们之间存在根本区别。当您查看 SQL 表并忽略约束时，所有列都属于同一类对象，而key and value in the PairwiseRDD有明确的含义。

回到你的问题来使用join你两者都需要key and value。可以说比使用干净得多0作为占位符将使用null单例，但确实没有办法解决它。

对于小数据，您可以以类似于广播连接的方式使用过滤器：

val moviesidBD = sc.broadcast(
  lines.map(x => x.split("\t")).map(_.head).collect.toSet)

movienames.filter{case (id, _) => moviesidBD.value contains id}

但如果你真的想要 SQL 式的连接，那么你应该简单地使用 SparkSQL。

val movieIdsDf = lines
   .map(x => x.split("\t"))
   .map(a => Tuple1(a.head))
   .toDF("id")

val movienamesDf = movienames.toDF("id", "name")

// Add optional join type qualifier 
movienamesDf.join(movieIdsDf, movieIdsDf("id") <=> movienamesDf("id"))