Scala spark 如何与 List[Option[Map[String, DataFrame]]] 交互

Posted 2023-03-16

【中文标题】Scala spark 如何与 List[Option[Map[String, DataFrame]]] 交互

【英文标题】：Scala spark how to interact with a List[Option[Map[String, DataFrame]]]

【发布时间】：2021-04-04 21:53:29

【问题描述】：

我正在尝试与此 List[Option[Map[String, DataFrame]]] 进行交互，但遇到了一些麻烦。

里面有这样的东西：

customer1 -> dataframeX 
customer2 -> dataframeY 
customer3 -> dataframeZ

其中客户是一个标识符，将成为一个新列。

我需要合并 dataframeX、dataframeY 和 dataframeZ（所有 df 都有相同的列）。在我有这个之前：

map(_.get).reduce(_ union _).select(columns:_*)

它工作正常，因为我只有一个 List[Option[DataFrame]] 并且不需要标识符，但我在使用新列表时遇到了问题。我的想法是修改我的旧映射，我知道我可以执行“(0).get”之类的操作，这会给我带来“Map(customer1 -> dataframeX)”，但我不太确定如何在映射并获得最终的数据帧，它是所有三个加上标识符的并集。我的想法：

map(/*get identifier here along with dataframe*/).reduce(_ union _).select(identifier +: columns:_*)

最终结果会是这样的：

-------------------------------
|identifier | product  |State | 
-------------------------------
|  customer1|  prod1   |  VA  |
|  customer1|  prod132 |  VA  |
|  customer2|  prod32  |  CA  | 
|  customer2|  prod51  |  CA  |
|  customer2|  prod21  |  AL  |
|  customer2|  prod52  |  AL  |
-------------------------------

【参考方案1】：

您可以使用collect 将Option[Map[String, Dataframe]] 取消嵌套到Map[String, DataFrame]。要将标识符放入列中，您应该使用withColumn。所以你的代码可能看起来像：

import org.apache.spark.sql.functions.lit

val result: DataFrame = frames.collect 
    case Some(m) =>
      m.map 
        case (identifier, dataframe) => dataframe.withColumn("identifier", lit(identifier))
      .reduce(_ union _)
  .reduce(_ union _)

【参考方案2】：

大概是这样的吧？

list
  .flatten 
  .flatMap  
    _.map  case (id, df) => 
      df.withColumn("identifier", id)  
  .reduce(_ union _)