Eu estou tentando fazer alguma análise sobre conjuntos. Eu tenho um conjunto de dados de exemplo que se parece com isso:

orders.json

{"items":[1,2,3,4,5]}
{"items":[1,2,5]}
{"items":[1,3,5]}
{"items":[3,4,5]}

Tudo o que existe é um único campo que é uma lista de números que representam IDs.

Aqui está o script Spark que estou tentando executar:

val sparkConf = new SparkConf()
  .setMaster("local[*]")
  .setAppName("Dataframe Test")

val sc = new SparkContext(sparkConf)
val sql = new SQLContext(sc)

val dataframe = sql.read.json("orders.json")

val expanded = dataframe
  .explode[::[Long], Long]("items", "item1")(row => row)
  .explode[::[Long], Long]("items", "item2")(row => row)

val grouped = expanded
  .where(expanded("item1") !== expanded("item2"))
  .groupBy("item1", "item2")
  .count()

val recs = grouped
  .groupBy("item1")

Criandoexpanded egrouped é bom, em poucas palavrasexpanded é uma lista de todos os conjuntos possíveis de dois IDs em que os dois estavam no mesmo conjunto original.grouped filtra os IDs correspondentes e depois agrupa todos os pares exclusivos de IDs e produz uma contagem para cada um. O esquema e a amostra de dados degrouped estão:

root
 |-- item1: long (nullable = true)
 |-- item2: long (nullable = true)
 |-- count: long (nullable = false)

[1,2,2]
[1,3,2]
[1,4,1]
[1,5,3]
[2,1,2]
[2,3,1]
[2,4,1]
[2,5,2]
...

Então, minha pergunta é: como agora agrupo o primeiro item de cada resultado para ter uma lista de tuplas? Para os dados de exemplo acima, eu esperaria algo semelhante a este:

[1, [(2, 2), (3, 2), (4, 1), (5, 3)]]
[2, [(1, 2), (3, 1), (4, 1), (5, 2)]]

Como você pode ver no meu script comrecs, Pensei que você começaria fazendo um groupBy em 'item1', que é o primeiro item em cada linha. Mas depois disso, você fica com esse objeto GroupedData que possui ações muito limitadas. Realmente, você só pode fazer agregações como soma, média, etc. Eu só quero listar as tuplas de cada resultado.

Eu poderia facilmente usar funções RDD neste momento, mas isso parte do uso de Dataframes. Existe uma maneira de fazer isso com as funções do quadro de dados.