Spark 数据框 1.6

Posted 2023-04-15

【中文标题】Spark 数据框 1.6

【英文标题】：Spark Data frame 1.6

【发布时间】：2018-11-13 18:13:50

【问题描述】：

数据转储

Work_Id,Assigned_to,Date,Status   
R1,John,3/4/15,Not Started   
R1,John,3/5/15,In Progress        
R1,John,3/6/15,Finished     
R3,Alaxender,3/7/15,In Progress   
R3,Alaxender,3/8/15,In Progress   
R4,Patrick,3/9/15,Finished   
R5,Peter,3/11/15,Finished   
R7,George,3/13/15,Not Started   
R7,George,3/14/15,In Progress   
R8,John,3/15/15,In Progress    
R8,John,3/16/15,In Progress   
R9,Alaxender,3/17/15,Not Started

最终输出

Work_Id,Assigned_to,Date,Status   
R1,John,3/6/15,Finished    
R7,George,3/14/15,In Progress    
R9,Alaxender,3/17/15,Not Started    
R3,Alaxender,3/7/15,In Progress    
R3,Alaxender,3/8/15,In Progress    
R4,Patrick,3/9/15,Finished    
R5,Peter,3/11/15,Finished    
R8,John,3/15/15,In Progress    
R8,John,3/16/15,In Progress 

有一个与上面相同的数据转储，其中包含工作订单。如果对同一个人有后续请求且状态为“未开始”，则最后一条记录（按日期排序）将合格。如果只有一条状态为“未开始”的记录，则该记录为合格。

例如：

R1,John,3/4/15,Not Started    
R1,John,3/5/15,In Progress   
R1,John,3/6/15,Finished   

此记录将合格

R1,John,3/6/15,Finished

对于同一个人，除了状态为“未开始”之外的所有记录都将在输出中合格。

任何帮助将不胜感激，使用 scala 在 Spark 1.6 数据帧中完成此操作。

【问题讨论】：

这与您发布的最后一个问题有什么不同，但您自己没有尝试解决问题？ ***.com/questions/49472718/… ... idownvotedbecau.se/noattempt

感谢您对此的帮助..

@cricket_007，我们无法通过 collect_list 实现这一目标。完全需要一种不同的方法。

好吧，好吧，仍然......你能不能请edit 用一些 Scala 代码来回答你的问题？

@cricket_007，我在这里更新了答案。有没有更好的方法来做到这一点？

【参考方案1】：

我有一个答案，但这目前降低了工作绩效。有没有更好的方法来做到这一点？

val df = myFile.toDF()

val dfFilter = df.filter($"status" === "Not Started")

val dfSelect = dfFilter.select(($"Assigned_to").alias("person"))

val dfInner = dfSelect.join(df, $"person" === $"Assigned_to")

val windowSpec = Window.partitionBy($"Assigned_to").orderBy(col("Date").desc)

val dfRank = dfInner.withColumn("rank", rank().over(windowSpec)).filter($"rank" === "1")

val dfDrop = dfRank.drop($"rank").drop($"person")

val dfLeftOuter = df.join(dfSelect, $"Assigned_to" === $"person", "leftouter")

val nullDf = dfLeftOuter.filter($"person".isNull).drop($"person")

nullDf.unionAll(dfDrop).show