代码之家 › 专栏 › 技术社区 › maximodesousadias

如何根据条件删除日期后的记录

apache-spark-sql pyspark apache-spark

maximodesousadias · 技术社区 · 10 月前

我正在寻找一种优雅的方法来删除DataFrame中发生在最新事件之前的所有记录 '测试组件' 存在 “不满意” ,基于他们 'TEST_DT' 每个ID的值。

例如,给定ID 5000的以下DataFrame:

| ID   | TEST_ID | TEST_COMPONENT | TEST_DT                       |
|------|---------|----------------|-------------------------------|
| 5000 | ENGL    | SATISFACTORY   | 2023-01-04T00:00:00.000+11:00 |
| 5000 | ENGL    | SATISFACTORY   | 2022-09-07T00:00:00.000+10:00 |
| 5000 | OTHER   | NONE           | 2022-09-07T00:00:00.000+10:00 |
| 5000 | ENGL    | UNSATISFACTORY | 2016-05-23T00:00:00.000+10:00 |
| 5000 | OTHER   | NONE           | 2016-05-23T00:00:00.000+10:00 |
| 5000 | OTHER   | NONE           | 2016-05-23T00:00:00.000+10:00 |
| 5000 | OTHER   | NONE           | 2016-02-09T00:00:00.000+11:00 |
| 5000 | OTHER   | NONE           | 2016-02-09T00:00:00.000+11:00 |
| 5000 | OTHER   | NONE           | 2016-02-09T00:00:00.000+11:00 |
| 5000 | ENGL    | UNSATISFACTORY | 2014-05-29T00:00:00.000+10:00 |
| 5000 | OTHER   | NONE           | 2013-09-27T00:00:00.000+10:00 |

我只想保留最新的行 “不满意” 记录向前。此示例的期望输出为:

| ID   | TEST_ID | TEST_COMPONENT | TEST_DT                       |
|------|---------|----------------|-------------------------------|
| 5000 | ENGL    | SATISFACTORY   | 2023-01-04T00:00:00.000+11:00 |
| 5000 | ENGL    | SATISFACTORY   | 2022-09-07T00:00:00.000+10:00 |
| 5000 | OTHER   | NONE           | 2022-09-07T00:00:00.000+10:00 |
| 5000 | ENGL    | UNSATISFACTORY | 2016-05-23T00:00:00.000+10:00 |

我怎样才能有效地利用 PySpark ?

2 回复 | 直到 10 月前

ManishPrajapati 10 月前

以下是我使用Window函数的尝试:

window_spec = Window.partitionBy("ID")

df = df.withColumn("MAX_DATE",F.max(F.when(df['TEST_COMPONENT']=='UNSATISFACTORY',df['TEST_DT']).otherwise(None)).over(window_spec))
df_drop = df.filter((df['TEST_DT']>F.col('MAX_DATE')) | ((df['TEST_DT']==F.col('MAX_DATE')) & (df['TEST_COMPONENT']=='UNSATISFACTORY')))

df_drop.show(truncate=False)

maximodesousadias 10 月前

我通过在PySpark中执行以下3个步骤实现了这一点:

df_max = df.filter(col("TEST_COMPONENT")=="UNSATISFACTORY")\
.groupBy("ID")\
.agg(max("TEST_DT")\
.alias("LATEST_UNSAT"))

df = df.alias("a").join(\
  df_max.alias("b")\
  ,df.ID == df_max.ID\
  ,"left")

df = df.filter(col("TEST_DT") >= col("LATEST_UNSAT"))

这种方法是有效的,但我愿意听取更好的想法。

推荐文章

srinath tripuraneni · {DataFrameWriter CSV到HDFS文件系统}不分区写入数据

3 年前

Calcutta · Google Colab中的Spark SQL在大数据上失败

3 年前

Palkin Jangra · 使用循环在Pyspark数组元素上和元素本身内迭代两次

3 年前

Doraemon · PySpark:使用不同值的字符串类型列创建聚合列

3 年前

amol visave · spark作业失败时会发生什么?

3 年前

Alex Jolly · 如何在另一个pyspark数据帧中查询开始时间和结束时间之间的日期时间

3 年前

chun · pyspark dataframe在s3中两次写入csv文件

3 年前

Mod_x · 如何将特定列的行标题更改为行标题,并在pySpark中生成矩阵?

3 年前

katty · 动态地将参数传递给scala中的函数

7 年前

Abhishek Choudhary · 如何更新数组列?

7 年前