Spark Dataframe со сложными и вложенными данными - PullRequest
Новая
       45

Spark Dataframe со сложными и вложенными данными

1 голос
/ 19 апреля 2019

У меня есть 3 кадра данных. Назовите их dfA, dfB и dfC

У dfA есть 3 столбца

| Id |Имя |Age |

dfB сказал, что 5 столбцов.2-й столбец - это ссылка FK на запись dFA.

| Id |AId |Улица |Город |Zip |

Точно так же dfC имеет 3 столбца, также со ссылкой на dfA

| Id |AId |SomeField |

Используя Spark SQL, я могу выполнить JOIN через 3 

%sql

SELECT * FROM dfA
INNER JOIN dfB ON dfA.Id = dfB.AId
INNER JOIN dfC ON dfA.Id = dfC.AId

И я получу свой набор результатов, но он был «сведен», как SQL будет делать с табличными результатами, такими какthis.

Я хочу загрузить его в сложную схему, подобную этой 

val destinationSchema = new StructType()
  .add("id", IntegerType)
  .add("name", StringType)
  .add("age", StringType)
  .add("b", 
       new StructType()
        .add("street", DoubleType, true)
        .add("city", StringType, true)
    .add("zip", StringType, true)
      )
  .add("c",
       new StructType()
        .add("somefield", StringType, true)
      )

Есть идеи, как получить результаты SELECT и сохранить их в dataframe с указанием схемы?

В конечном итоге я хочу сохранить сложный StructType или JSON и загрузить его в базу данных Mongo с помощью Mongo Spark Connector.

Или есть ли лучший способ сделать это из 3отдельные кадры данных (которые изначально были 3 отдельными CSV-файлами, которые были прочитаны)?

Ответы [ 2 ]

1 голос
/ 19 апреля 2019

предыдущий работал, если все записи имели отношение 1: 1.

вот как вы можете добиться этого за 1: M (подсказка: используйте collect_set для группировки строк) 

import org.apache.spark.sql.functions._

val destDF = atableDF
  .join(btableDF, atableDF("id") === btableDF("id")).drop(btableDF("id"))
  .join(ctableDF, atableDF("id") === ctableDF("id")).drop(ctableDF("id"))
  .groupBy($"id",$"name",$"age")
  .agg(collect_set(struct($"street",$"city",$"zip")) as "b",collect_set(struct($"somefield")) as "c")

val jsonDestDF = destDF.select(to_json(struct($"*")).as("row"))

display(jsonDestDF)

, который даст вам следующий вывод: 

row
"{""id"":102,""name"":""Damian"",""age"":""23"",""b"":[{""street"":""Short Street"",""city"":""New York"",""zip"":""70701""}],""c"":[{""somefield"":""pears""},{""somefield"":""pineapples""}]}"
"{""id"":100,""name"":""John"",""age"":""43"",""b"":[{""street"":""Dark Road"",""city"":""Washington"",""zip"":""98002""}],""c"":[{""somefield"":""appples""}]}"
"{""id"":101,""name"":""Sally"",""age"":""34"",""b"":[{""street"":""Light Ave"",""city"":""Los Angeles"",""zip"":""90210""}],""c"":[{""somefield"":""grapes""},{""somefield"":""peaches""},{""somefield"":""bananas""}]}"

пример данных, которые я использовал на тот случай, если кто-то захочет сыграть:

atable.csv 

100,"John",43
101,"Sally",34
102,"Damian",23
104,"Rita",14
105,"Mohit",23

btable.csv: 

100,"Dark Road","Washington",98002
101,"Light Ave","Los Angeles",90210
102,"Short Street","New York",70701
104,"Long Drive","Buffalo",80345
105,"Circular Quay","Orlando",65403

ctable.csv: 

100,"appples"
101,"bananas"
102,"pears"
101,"grapes"
102,"pineapples"
101,"peaches"
1 голос
/ 19 апреля 2019

с учетом трех фреймов данных, загруженных из CSV-файлов, вы можете сделать это: 

import org.apache.spark.sql.functions._

val destDF = atableDF
  .join(btableDF, atableDF("id") === btableDF("id")).drop(btableDF("id"))
  .join(ctableDF, atableDF("id") === ctableDF("id")).drop(ctableDF("id"))
  .select($"id",$"name",$"age",struct($"street",$"city",$"zip") as "b",struct($"somefield") as "c")

val jsonDestDF = destDF.select(to_json(struct($"*")).as("row"))

, что приведет к выводу: 

row
"{""id"":100,""name"":""John"",""age"":""43"",""b"":{""street"":""Dark Road"",""city"":""Washington"",""zip"":""98002""},""c"":{""somefield"":""appples""}}"
"{""id"":101,""name"":""Sally"",""age"":""34"",""b"":{""street"":""Light Ave"",""city"":""Los Angeles"",""zip"":""90210""},""c"":{""somefield"":""bananas""}}"
"{""id"":102,""name"":""Damian"",""age"":""23"",""b"":{""street"":""Short Street"",""city"":""New York"",""zip"":""70701""},""c"":{""somefield"":""pears""}}"
...