कैसे आबाद करने के लिए नक्शा[स्ट्रिंग,Dataframe] के रूप में एक कॉलम में एक Dataframe स्काला में

आप पूछ रहे हैं उत्पन्न करने के लिए एक पंक्ति प्रति dataframe. सावधान रहो, अगर एक dataframes पर्याप्त इतनी बड़ी है कि यह निहित नहीं किया जा सकता है, जहां एक प्रबंधक, इस कोड को तोड़ देंगे ।

की पहली उत्पन्न डेटा और नक्शे dfs के प्रकार 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)

तो, प्रत्येक के लिए dataframe का नक्शा, हम उत्पन्न दो कॉलम है. big_map सहयोगियों के प्रत्येक स्तंभ के नाम dataframe करने के लिए अपने मूल्य (कास्ट में स्ट्रिंग के लिए एक सुसंगत प्रकार). df बस का नाम शामिल है dataframe. हम तो यूनियन के सभी dataframes के साथ reduce और समूह द्वारा name (यह हिस्सा है जहां हर एक dataframe समाप्त होता है पूरी तरह से एक पंक्ति में है, और इसलिए एक एक प्रबंधक).

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

यहाँ एक समाधान है विशिष्ट उपयोग के मामले:

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

स्पार्क सत्र बनाने के लिए आवश्यक है खाली DataFrame संचायक पर गुना करने के लिए.

वैकल्पिक रूप से अगर आप की गारंटी कर सकते हैं Map है गैर खाली है.

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