Spark- SQL с использованием withColumnRenamed ()

Question

Я пытаюсь загрузить файл Parquet со столбцами storyId1 и publisher1. Я хочу найти все пары издателей, которые публикуют sh статей об одних и тех же историях. Для каждой пары издателей необходимо сообщить количество совместно опубликованных историй. Где совместно опубликованная история в истории, опубликованной обоими издателями. Сообщите пары в порядке убывания частоты. Решение должно соответствовать следующим правилам: 1. Не должно быть никаких реплицируемых записей, например: NASDAQ, NASDAQ, 1000 2. Одна и та же пара не должна встречаться дважды в противоположном порядке. Должно произойти только одно из следующего: NASDAQ, Reuters, 1000 Reuters, NASDAQ, 1000 (т. Е. Неверно иметь обе вышеперечисленные строки в вашем результате)

Теперь он попробовал следующий код:

> import org.apache.spark.sql._
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions._
import spark.implicits._


val worddocDF = spark.read.parquet("file:///home/user204943816622/t4_story_publishers.parquet")
val worddocDF1 = spark.read.parquet("file:///home/user204943816622/t4_story_publishers.parquet")
worddocDF.cache()
val joinDF = worddocDF.join(worddocDF1, "storyId1").withColumnRenamed("worddocDF.publisher1", "publisher2")
joinDF.filter($"publisher1" !== $"publisher2")

Формат ввода:

[ddUyU0VZz0BRneMioxUPQVP6sIxvM, Livemint]

[ddUyU0VZz0BRneMioxUPQVP6sIxvM, IFA Magazine]

[ddUyU0VZz0BRneMioxUPQVP6sIxvM, Moneynews]

[ddUyU0VZz0BRneMioxUPQVP6sIxvM, NASDAQ]

[dPhGU51DcrolUIMxbRm0InaHGA2XM, IFA Magazine]

[ddUyU0VZz0BRneMioxUPQVP6sIxvM, Los Angeles Times]

[dPhGU51DcrolUIMxbRm0InaHGA2XM, NASDAQ]

Требуемый вывод:

[

NASDAQ,IFA Magazine,2]

[Moneynews,Livemint,1]

[Moneynews,IFA Magazine,1]

[NASDAQ,Livemint,1]

[NASDAQ,Los Angeles Times,1]

[Moneynews,Los Angeles Times,1]

[Los Angeles Times,IFA Magazine,1]

[Livemint,IFA Magazine,1]

[NASDAQ,Moneynews,1]

[Los Angeles Times,Livemint,1]

Answer 1

import spark.implicits._

    wordDocDf.as("a")
    .join(
      wordDocDf.as("b"),
      $"a.storyId1" === $"b.storyId1" && $"a.publisher1" =!= $"b.publisher1",
      "inner"
    )
    .select(
      $"a.storyId1".as("storyId"),
      $"a.publisher1".as("publisher1"),
      $"b.publisher1".as("publisher2")
    )