Map[string,Dataframe] ' i scala'daki bir Dataframe'de sütun olarak nasıl doldurabilirim

Veri çerçevesi başına bir satır oluşturmak istiyorsunuz. Dikkatli olun, veri çerçevelerinden biri tek bir yürütücüde bulunamayacak kadar büyükse, bu kod kırılır.

Önce verileri ve haritayı oluşturalım dfs tür 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)

Ardından, haritanın her veri çerçevesi için iki sütun oluştururuz. big_map dataframe'in her sütun adını değeriyle ilişkilendirir (tutarlı bir türe sahip olmak için dizeye dökülür). df sadece veri çerçevesinin adını içerir. Sonra tüm dataframes Birliği biz reduce ve gruplandır name (bu, her bir veri çerçevesinin tamamen bir satırda ve dolayısıyla bir yürütücüde bittiği kısımdır).

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}]|
+---+-----------------------------------------------------------------------------------+

İşte kullanım durumunuza özgü bir çözüm:

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))

Kıvılcım oturumu, katlanacak boş DataFrame akümülatörünü oluşturmak için gereklidir.

Alternatif olarak, garanti edebilirseniz Map sigara boş.

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