Link entre objeto scala, conjunto de dados e dataframe [duplicado]

Question

Eu criei a seguinte classe de caso:

case class Data(ads:Option[Ads])

case class Ads(subject: Option[String]
           , body:Option[String]
           , price:Option[Int]
           , location:Option[Location]
           , attribut:Option[Seq[Attribut]]
 )

case class Location(city:Option[String]
                , zipcode:Option[String])

case class Attribut(key_label:Option[String]
                , value_label:Option[String]
)

E analiso um formato JSON (parte de um HTML) com estrutura de jogo.

Finalmente consigo um Object Ads

JsSuccess(Ads(Some("Subject"), SOme("Body"), Some(Price), Some(Location(Some("City"), Some("Zipcode")), Some(Attribut("key_label", "value_label")) 

Desejo salvar isso em um arquivo CSV da seguinte maneira:

Subject   Body           Price   City  Zipcode  Key_Label  Value_Label
Play      Playing games  532     Geneve 95      GEN        Gen2

Eu converto o objeto em uma Lista de Ads(Some("Subject"), Some("Body"), Some(Price), Some(Location(Some("City"), Some("Zipcode")), Some(Attribut("key_label", "value_label") e converto essa lista em um DataFrame.

Mas eu tenho apenas uma coluna Value que contém todos os elementos do objeto.

    Value
    (Some("Subject"), SOme("Body"), Some(Price), Some(Location(Some("City"), Some("Zipcode")), Some(Attribut("key_label", "value_label")

Alguém tem uma ideia, por favor? Eu realmente não entendo como vincular o objeto scala com conjunto de dados e dataframe. Obrigado pela ajuda.

Answer 1

Os comentários são úteis, mas a função de nivelamento genérico pode não gerar colunas na ordem desejada e / ou lidar com a colocação de elementos da matriz em suas próprias colunas separadas.

Supondo que seu arquivo JSON contenha linhas como:

{"ads": {"subject": "abc", "body": "doing something", "price": 13, "location": {"city": "Houston", "zipcode": 39014}, "attribut": [{"key_label": "a", "value_label": "b"}]}}

Se o arquivo for razoavelmente consistente e você já estiver incluindo o Spark como uma dependência, provavelmente não precisará usar uma biblioteca separada para analisar o JSON.

Você precisará usar a explodefunção para lidar com o fato de que a coluna "atribut" é uma lista. Use a explode_outerfunção ao invés, se as listas podem estar vazias, mas você gostaria de preservar os valores de outras colunas.

import org.apache.spark.sql.functions._
// assuming spark is the Spark Session
val df = spark.read.json("mydata.json")

val df1 = df.select(col("ads.subject").alias("Subject"), col("ads.body").alias("Body"),
          col("ads.location.city").alias("City"), col("ads.location.zipcode").alias("Zipcode"),
          explode(col("ads.attribut")))

val resultDF = df1.select(col("Subject"), col("Body"), col("City"), col("Zipcode"),
               col("col.key_label"), col("col.value_label"))

resultDF.show produziria:

+-------+---------------+-------+-------+---------+-----------+
|Subject|           Body|   City|Zipcode|key_label|value_label|
+-------+---------------+-------+-------+---------+-----------+
|    abc|doing something|Houston|  39014|        a|          b|
+-------+---------------+-------+-------+---------+-----------+

Para produzir como um único arquivo CSV no diretório especificado, com cabeçalhos:

resultDF.repartition(1).write.option("header", "true").csv("/tmp/my-output-dir/")