pyspark中的Groupby数据框和过滤器答案

【问题标题】：Groupby dataframe and filter in pysparkpyspark中的Groupby数据框和过滤器
【发布时间】：2021-09-24 17:01:07
【问题描述】：

以下是我的输入火花数据框，有人可以帮助我使用所需的数据框或至少方法吗？

unique-id	status
1	OAOS-STP
1	OAOS-nonSTP
1	manual
2	OAOS-nonSTP
2	manual
3	OAOS-STP
3	OAOS-nonSTP
4	OAOS-STP
4	manual

我期待的输出数据框：

unique-id	status
1	OAOS-STP
2	OAOS-nonSTP
3	OAOS-STP
4	OAOS-STP

OAOS-STP > OAOS-nonSTP > 手册按优先顺序排列。提前致谢。

【问题讨论】：

状态真的是文字吗？因为文本理解排序不是微不足道的……例如，“第四”介于“第一”和“第二”之间。如果您试图用虚拟数据过度简化问题，那并不总是一个好主意。您可能应该展示您的实际用例
@Steven，我改变了状态栏

标签： python apache-spark pyspark apache-spark-sql

【解决方案1】：

您可以使用字典和user-defined function 将每一行的状态链接到表示状态顺序的整数，然后在unique-id 上执行groupBy，聚合min 状态顺序，最后从该状态顺序中检索状态，如下：

from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType


status_dict = {'OAOS-STP': 1, 'OAOS-nonSTP': 2, 'manual': 3}


@F.udf(returnType=IntegerType())
def get_status_order(status):
    return status_dict.get(status)


data = input_df.withColumn('status_order', F.struct(get_status_order('status'), F.col('status'))) \
    .groupBy('unique-id') \
    .agg(F.min('status_order').alias('status')) \
    .withColumn('status', F.col('status.status')) \
    .orderBy('unique-id')  # Optional

【讨论】：

【解决方案2】：

您可以使用窗口函数并通过对状态值进行排序来为每个唯一键选择最高值。

PS：我在scala中编码

import spark.implicits._
import org.apache.spark.sql.functions._
val df = Seq(
  (1, "OAOS-STP"),
  (1, "OAOS-nonSTP"),
  (1, "manual"),
  (2, "OAOS-nonSTP"),
  (2, "manual"),
  (3, "OAOS-STP"),
  (3, "OAOS-nonSTP"),
  (4, "OAOS-STP"),
  (4, "manual")
).toDF("unique-id", "status")

import org.apache.spark.sql.expressions.Window
val df2 = df.withColumn("lower_status", lower($"status"))
val windowSpec  = Window.partitionBy("unique-id").orderBy("status")

val df3 = df2
.withColumn("rank", rank().over(windowSpec)).filter($"rank" === 1)
.drop("rank")
.drop("lower_status")

df3.show(false) 的输出将是

【讨论】：

嗨 Vikas，我已经更新了状态列，我可以像你一样将整数值分配给状态列，执行窗口函数然后删除值吗？你有什么建议？
@vik，对数值进行排序要容易得多，因为对于字符串值，字母（大写或小写）的变化会使事情变得困难。让我试试看它是否有效。
可以考虑全部小写
是的，这就是我要做的......它只是在 databricks 社区云上生成一个集群需要一些时间，所以等待它生成并测试逻辑
@vik，我已根据您的更新值更新了我的回复。如果实际数据集的状态值相同，请随意使用上述解决方案，但如果值超过此值，您可能需要调整上述逻辑或根据优先级将状态值映射为数值。