Как присоединиться к двум свечам RDD

Question

У меня есть 2 спарк-RDD, первый содержит отображение между некоторыми индексами и идентификаторами, которые являются строками, а второй содержит наборы связанных индексов

val ids = spark.sparkContext.parallelize(Array[(Int, String)](
      (1, "a"), (2, "b"), (3, "c"), (4, "d"), (5, "e"))).toDF("index", "idx")


val relationships = spark.sparkContext.parallelize(Array[(Int, Int)](
  (1, 3), (2, 3), (4, 5))).toDF("index1", "index2")

Я хочу как-то присоединиться к этим СДР (или слиянию, или sql, или к любой наилучшей искровой практике), чтобы вместо них иметь в конце связанные идентификаторы:

Результат моего комбинированного RDD должен вернуть:

("a", "c"), ("b", "c"), ("d", "e")

Любая идея, как я могу выполнить эту операцию оптимальным образом, не загружая ни один СДР в карту памяти (потому что в моих сценариях эти СДР могут потенциально загружать миллионы записей)

Answer 1

Вы можете подойти к этому, создав два views из DataFrame следующим образом

relationships.createOrReplaceTempView("relationships");
ids.createOrReplaceTempView("ids");

Затем выполните следующий запрос SQL, чтобы сгенерировать требуемый результат, который выполняет внутреннее соединение между представлениями relationships и ids, чтобы сгенерировать требуемый результат

import sqlContext.sql;
val result = spark.sql("""select t.index1, id.idx from 
                                (select id.idx as index1, rel.index2 
                               from relationships rel
                               inner join
                               ids id on rel.index1=id.index) t
                         inner join
                         ids id
                         on id.index=t.index2
                      """);

result.show()

Другой подход, использующий DataFrame без создания views

relationships.as("rel").
join(ids.as("ids"),  $"ids.index" === $"rel.index1").as("temp").
join(ids.as("ids"), $"temp.index2"===$"ids.index").
select($"temp.idx".as("index1"), $"ids.idx".as("index2")).show