scala – 如何在Spark中对GroupedData进行自定义操作？

发布时间：2020-12-16 09:24:54 所属栏目：安全来源：网络整理

导读：我想重写一些用RDD编写的代码来使用DataFrames.在我找到这个之前,它工作得很顺利： events .keyBy(row = (row.getServiceId + row.getClientCreateTimestamp + row.getClientId,row) ) .reduceByKey((e1,e2) = if(e1.getClientSendTimestamp = e2.getClientS

我想重写一些用RDD编写的代码来使用DataFrames.在我找到这个之前,它工作得很顺利：

events
  .keyBy(row => (row.getServiceId + row.getClientCreateTimestamp + row.getClientId,row) )
  .reduceByKey((e1,e2) => if(e1.getClientSendTimestamp <= e2.getClientSendTimestamp) e1 else e2)
  .values

它很简单

events
  .groupBy(events("service_id"),events("client_create_timestamp"),events("client_id"))

但下一步是什么？如果我想迭代当前组中的每个元素怎么办？它甚至可能吗？
提前致谢.

解决方法

GroupedData不能直接使用.数据没有物理分组,只是一个逻辑操作.你必须应用一些agg方法的变体,例如：

events
 .groupBy($"service_id",$"client_create_timestamp",$"client_id")
 .min("client_send_timestamp")

要么

events
 .groupBy($"service_id",$"client_id")
 .agg(min($"client_send_timestamp"))

其中client_send_timestamp是您要聚合的列.

如果您想保留信息而不是聚合只需加入或使用窗口功能 – 请参阅Find maximum row per group in Spark DataFrame

Spark还支持用户定义的聚合函数 – 请参阅How to define and use a User-Defined Aggregate Function in Spark SQL?

Spark 2.0

您可以使用Dataset.groupByKey将组公开为迭代器.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!