spark - 加入一对多关系数据框答案

【问题标题】：spark - join one to many relationship dataframesspark - 加入一对多关系数据框
【发布时间】：2016-11-09 03:38:44
【问题描述】：

让我们来解决以下玩具问题，我有以下案例类：

case class Order(id: String, name: String, status: String)
case class TruncatedOrder(id: String)
case class Org(name: String, ord: Seq[TruncatedOrder])

我现在有以下定义的变量

val ordersDF = Seq(Order("or1", "stuff", "shipped"), Order("or2", "thigns", "delivered") , Order("or3", "thingamabobs", "never received"), Order("or4", "???", "what?")).toDS()
val orgsDF = Seq(Org("tupper", Seq(TruncatedOrder("or1"), TruncatedOrder("or2"), TruncatedOrder("or3"))), Org("ware", Seq(TruncatedOrder("or3"), TruncatedOrder("or4")))).toDS()

例如，我想要一个如下所示的数据点
Ord("tupper", Array(Joined("or1", "stuff", "shipped"), Joined("or2", "things", "delivered"), ...)

我想知道如何格式化我的join 语句和过滤语句。

【问题讨论】：

您是否尝试将DF 保存为temp table 并编写查询？一旦您对此感到满意，您就可以使用 DataFrame API 进行转换。
@Shankar 我没想到，你怎么看那个临时表被使用了？
你可以将df存储到temple ordersdf.registerTempTable("orders")，然后使用sqlContext.sql("write sql query with join and where condition")
你可以看看这个spark.apache.org/docs/2.0.0/…

标签： apache-spark

【解决方案1】：

以下是我如何将数据转换为我想要的格式。这个答案深受@ulrich 和@Mariusz 提供的答案的启发。

val ud = udf((col: String, name: String, status: String) => { Seq(col, name, status)})

orgsDF
  .select($"name".as("ordName"),explode($"ord.id"))
  .join(ordersDF, $"col" === $"id").drop($"id")
  .select($"ordName", ud($"col", $"name", $"status"))
  .groupBy($"ordName")
  .agg(collect_set($"order"))
  .show()

    +-------+--------------------------------------------------------------------------------------------------------------------------+
    |ordName|orders                                                                                                                    |
    +-------+--------------------------------------------------------------------------------------------------------------------------+
    |ware   |[WrappedArray(or4, ???, what?), WrappedArray(or3, thingamabobs, never received)]                                          |
    |tupper |[WrappedArray(or1, stuff, shipped), WrappedArray(or2, thigns, delivered), WrappedArray(or3, thingamabobs, never received)]|
    +-------+--------------------------------------------------------------------------------------------------------------------------+

【讨论】：

【解决方案2】：

这个怎么样？

spark.conf.set("HiveSupport.enabled", true)

orgsDF.select('name,explode('ord))
      .map {case row: Row =>(row(0).toString,row(1).toString.filterNot("[]()".contains(_))) }.toDF("name",("ord"))
      .join(ordersDF.select('id,'status,'name.as("name2") ),'ord === 'id).drop("id")
      .select('name,concat('ord, lit(","),'Status, lit(","),'name2 ).as("info"))
      .groupBy('name) 
      .agg(collect_set('info))
      .show()

+------+--------------------+
|  name|   collect_set(info)|
+------+--------------------+
|  ware|[[or3,never recei...|
|tupper|[[or1,shipped,stu...|
+------+--------------------+

【讨论】：

【解决方案3】：

如果您遵循以下步骤，您可以轻松编写一对多：

使用explode 展平orgsDF
执行标准内连接
使用collect_list 按tuppler-thing 聚合

【讨论】：

两个集合的大小都接近 1tb，我不太确定 collect 有什么可能。
collect_list 是一个聚合函数，与collect() 终端操作相反，它不会向驱动程序发送任何结果。试试看！