Присоединиться к двум RDD в искры

Question

У меня есть два 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) и 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 работает, так же, как в Спарк могут присоединиться по клавишам как из РДД-х.

Answer 1

Операция присоединения определяется только на PairwiseRDDs, которые сильно отличаются от отношения/таблицы в SQL. Каждый элемент PairwiseRDD представляет собой Tuple2, где первым элементом является key, а второй - value. Оба могут содержать сложные объекты, пока key обеспечивает содержательную hashCode

Если вы хотите, чтобы думать об этом в SQL-иш можно рассматривать как ключ все, что происходит в ON пункта и value содержит выбранные столбцы.

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

Хотя эти подходы выглядят схожими на первый взгляд, и вы можете выразить их с помощью другого, существует одно фундаментальное отличие. Когда вы смотрите на таблицу SQL и игнорируете ограничения, все столбцы принадлежат одному классу объектов, а key и value в PairwiseRDD имеют четкое значение.

Возвращаясь к вашей проблеме, чтобы использовать join, вам нужны как key, так и value. Вероятно, гораздо чище, чем использование 0 в качестве заполнителя, будет использовать null singleton, но на самом деле его нет.

Для небольших данных, которые вы можете использовать фильтр подобным образом вещать присоединиться:

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"))