Ako naplniť Mapa[string,Dataframe] ako stĺpca v Dataframe v scala

Question 1

Mám Map[String, Dataframe]. Som chcete skombinovať všetky údaje, ktoré sú vo vnútri, že Mapu do jedného Dataframe. Môže dataframe majú stĺpec Mapu typ?

def sample(dfs : Map[String,Dataframe]): Dataframe =
{
.........
}

Príklad:

DF1

id name age
1  aaa  23
2  bbb  34

DF2

game  time  score
ludo  10    20
rummy 30    40

Som prejsť viac ako dva DFs ako Mapu na funkciu. Potom dal údajov z každej dataframes do jedného stĺpca výstup dataframe ako formáte json.

z DF

+---------------------------------------------------------------------------------------+
| column1                                                                              |
+---------------------------------------------------------------------------------------+
| [{"id":"1","name":"aaa","age":"23"},{"id":21","name":"bbb","age":"24"}]               |
| [{"game":"ludo","time":"10","score":"20"},{"game":"rummy","time":"30","score":"40"}]  |
+---------------------------------------------------------------------------------------+

Question 2

Môžete sa pýtať na generovanie jeden riadok na dataframe. Buďte opatrní, ak jeden z dataframes je dostatočne veľké na to, aby nemohol byť obsiahnuté v jednej exekútor, tento kód bude prestávka.

Poďme najprv vytvoriť údaje a mapa dfs druh Map[String, DataFrame].

val df1 = Seq((1, "aaa", 23), (2, "bbb", 34)).toDF("id", "name", "age")
val df2 = Seq(("ludo", 10, 20), ("rummy", 10, 40)).toDF("game", "time", "score")
dfs = Seq(df1, df2)

Potom, pre každé dataframe mapy, generujú dva stĺpce. big_map združuje každý stĺpec názov dataframe na jeho hodnotu (cast v reťazec mať jednotné typ). df jednoducho obsahuje názov dataframe. Potom sme únie všetky dataframes s reduce a skupiny name (toto je časť, kde každý dataframe skončí úplne v jednom riadku, a preto jednou z jednej exekútor).

dfs
    .toSeq
    .map{ case (name, df) => df
        .select(map(
             df.columns.flatMap(c => Seq(lit(c), col(c).cast("string"))) : _*
        ) as "big_map")
        .withColumn("df", lit(name))}
    .reduce(_ union _)
    .groupBy("df")
    .agg(collect_list('big_map) as "column1")
    .show(false)

+---+-----------------------------------------------------------------------------------+
|df |column1                                                                            |
+---+-----------------------------------------------------------------------------------+
|df0|[{id -> 1, name -> aaa, age -> 23}, {id -> 2, name -> bbb, age -> 34}]             |
|df1|[{game -> ludo, time -> 10, score -> 20}, {game -> rummy, time -> 10, score -> 40}]|
+---+-----------------------------------------------------------------------------------+

Question 3

Tu je riešenie špecifické pre vaše použitie-prípad:

import org.apache.spark.sql._

def sample(dfs : Map[String, DataFrame])(implicit spark: SparkSession): DataFrame =
  dfs
    .values
    .foldLeft(spark.emptyDataFrame)((acc, df) => acc.union(df))

Iskra relácie je potrebné vytvoriť prázdny DataFrame akumulátor zložiť na.

Prípadne ak môžete záruka Map nie je prázdny.

def sample(dfs : Map[String, DataFrame]): DataFrame =
  dfs
    .values
    .reduce((acc, df) => acc.union(df))

Oli · Answer 1 · 2021-11-24T07:05:52

Môžete sa pýtať na generovanie jeden riadok na dataframe. Buďte opatrní, ak jeden z dataframes je dostatočne veľké na to, aby nemohol byť obsiahnuté v jednej exekútor, tento kód bude prestávka.

Poďme najprv vytvoriť údaje a mapa dfs druh Map[String, DataFrame].

val df1 = Seq((1, "aaa", 23), (2, "bbb", 34)).toDF("id", "name", "age")
val df2 = Seq(("ludo", 10, 20), ("rummy", 10, 40)).toDF("game", "time", "score")
dfs = Seq(df1, df2)

Potom, pre každé dataframe mapy, generujú dva stĺpce. big_map združuje každý stĺpec názov dataframe na jeho hodnotu (cast v reťazec mať jednotné typ). df jednoducho obsahuje názov dataframe. Potom sme únie všetky dataframes s reduce a skupiny name (toto je časť, kde každý dataframe skončí úplne v jednom riadku, a preto jednou z jednej exekútor).

dfs
    .toSeq
    .map{ case (name, df) => df
        .select(map(
             df.columns.flatMap(c => Seq(lit(c), col(c).cast("string"))) : _*
        ) as "big_map")
        .withColumn("df", lit(name))}
    .reduce(_ union _)
    .groupBy("df")
    .agg(collect_list('big_map) as "column1")
    .show(false)

+---+-----------------------------------------------------------------------------------+
|df |column1                                                                            |
+---+-----------------------------------------------------------------------------------+
|df0|[{id -> 1, name -> aaa, age -> 23}, {id -> 2, name -> bbb, age -> 34}]             |
|df1|[{game -> ludo, time -> 10, score -> 20}, {game -> rummy, time -> 10, score -> 40}]|
+---+-----------------------------------------------------------------------------------+

gatear · Answer 2 · 2021-11-23T14:30:01

Tu je riešenie špecifické pre vaše použitie-prípad:

import org.apache.spark.sql._

def sample(dfs : Map[String, DataFrame])(implicit spark: SparkSession): DataFrame =
  dfs
    .values
    .foldLeft(spark.emptyDataFrame)((acc, df) => acc.union(df))

Iskra relácie je potrebné vytvoriť prázdny DataFrame akumulátor zložiť na.

Prípadne ak môžete záruka Map nie je prázdny.

def sample(dfs : Map[String, DataFrame]): DataFrame =
  dfs
    .values
    .reduce((acc, df) => acc.union(df))

môžete mi prosím skontrolujte otázku som pridal príklad teraz.Každý vstup dataframe, ktoré som si má diff schém, tak chcem, údaje z celého textu dataframe byť obývané ako stĺpci ,takže môj výstup dataframe má údajov každého vstupu dataframe v jednom stĺpci

Ako naplniť Mapa[string,Dataframe] ako stĺpca v Dataframe v scala

Otázka

Najlepšiu odpoveď

V iných jazykoch

Táto stránka je v iných jazykoch

Populárne v tejto kategórii

Obľúbené otázky v tejto kategórii