从数据集中同时聚合多个字段

Question

我有以下方案的数据：

sourceip
destinationip
packets sent

我想从这些数据中计算出几个聚合字段，并具有以下模式：

ip 
packets sent as sourceip
packets sent as destination

在RDD的快乐日子里，我可以使用聚合，定义{ip-的映射

在Dataset/Dataframe聚合中不再可用，而是可以使用UDAF，不幸的是，从我使用UDAF的经验来看，它们是不可变的，这意味着它们不能使用（必须在每次映射更新时创建一个新实例）示例解释在这里

一方面，从技术上讲，我可以将数据集转换为RDD、聚合等，然后返回数据集。我预计这会导致性能下降，因为数据集更加优化。由于复制，UDAF是不可能的。

是否有其他方法来执行聚合？

Answer 1

这是一个使用分解的pyspark版本。它更加冗长，但逻辑与flatMap版本完全相同，仅使用纯数据帧代码。

sc\
  .parallelize([("ip1", "ip2", 5), ("ip2", "ip3", 7), ("ip2", "ip1", 1), ("ip3", "ip2", 3)])\
  .toDF(("from", "to", "p"))\
  .select(F.explode(F.array(\
      F.struct(F.col("from").alias("ip"),\
               F.col("p").alias("received"),\
               F.lit(0).cast("long").alias("sent")),\
      F.struct(F.col("to").alias("ip"),\
               F.lit(0).cast("long").alias("received"),\
               F.col("p").alias("sent")))))\
  .groupBy("col.ip")\
  .agg(F.sum(F.col("col.received")).alias("received"), F.sum(F.col("col.sent")).alias("sent"))

// +---+----+--------+
// | ip|sent|received|
// +---+----+--------+
// |ip2|   8|       8|
// |ip3|   3|       7|
// |ip1|   5|       1|
// +---+----+--------+

Answer 2

在没有任何自定义聚合的情况下进行它的一种方法是使用平面图（或爆炸用于数据帧），如下所示：

case class Info(ip : String, sent : Int, received : Int)
case class Message(from : String, to : String, p : Int)
val ds = Seq(Message("ip1", "ip2", 5), 
             Message("ip2", "ip3", 7), 
             Message("ip2", "ip1", 1), 
             Message("ip3", "ip2", 3)).toDS()

ds
    .flatMap(x => Seq(Info(x.from, x.p, 0), Info(x.to, 0, x.p)))
    .groupBy("ip")
    .agg(sum('sent) as "sent", sum('received) as "received")
    .show


// +---+----+--------+
// | ip|sent|received|
// +---+----+--------+
// |ip2|   8|       8|
// |ip3|   3|       7|
// |ip1|   5|       1|
// +---+----+--------+

就性能而言，我不确定与自定义聚合相比，flatMap是否是一种改进。

Answer 3

听起来您需要一个标准的熔化（如何熔化Spark数据帧？）和枢轴组合：

val df = Seq(
  ("192.168.1.102", "192.168.1.122", 10),
  ("192.168.1.122", "192.168.1.65", 10),
  ("192.168.1.102", "192.168.1.97", 10)
).toDF("sourceip", "destinationip", "packets sent")


df.melt(Seq("packets sent"), Seq("sourceip", "destinationip"), "type", "ip")
  .groupBy("ip")
  .pivot("type", Seq("sourceip", "destinationip"))
  .sum("packets sent").na.fill(0).show

// +-------------+--------+-------------+             
// |           ip|sourceip|destinationip|
// +-------------+--------+-------------+
// | 192.168.1.65|       0|           10|
// |192.168.1.102|      20|            0|
// |192.168.1.122|      10|           10|
// | 192.168.1.97|       0|           10|
// +-------------+--------+-------------+