PySpark Groupby条件过滤

问题描述

我正在研究pyspark中的结构化数据框。从s3读取实木复合地板格式的数据。然后，我想用条件过滤掉一些数据。 例如：

Data:
key_1  value rec_date
  A     1    2020-01-01
  A     2    2020-01-02
  A    10    2020-01-03
  B    10    2020-10-10
  B    10    2020-10-11
  B    10    2020-10-12

Filter Condition:
{"A":{'abnormal_daterange':[('2020-01-01','2020-01-02'),('2020-02-01','2020-02-04')]},"B": {'abnormal_daterange':[('2020-10-10','2020-10-11')]}
}

Expected result:
key_1  value rec_date
  A    10    2020-01-03
  B    10    2020-10-12

我知道我可以使用@pandas-udf来创建要group by的函数并根据条件进行过滤，但是我想简化代码并且不要过度设计。

任何结构化数据帧filter /内置函数都可以有效过滤具有条件的数据吗？

解决方法

假设您有df和filterDf：

df = spark.createDataFrame(
    [
        ("A","1","2020-01-01"),("A","2","2020-01-02"),"10","2020-01-03"),("B","2020-10-10"),"2020-10-11"),"2020-10-12")
    ],['key_1','value','rec_date']
)

filterDf = spark.createDataFrame(
    [
        ("A","2020-01-01","2020-10-01","2020-10-11")
    ],["key","start_date","end_date"]
)

然后，您可以使用左反连接对df的内容过滤filterDf：

res = df.join(
    filterDf.hint("broadcast"),(col("key_1") == col("key")) & (col("rec_date").between(col("start_date"),col("end_date"))),"leftanti"
)

结果将是：

+-----+-----+----------+
|key_1|value|  rec_date|
+-----+-----+----------+
|    A|   10|2020-01-03|
|    B|   10|2020-10-12|
+-----+-----+----------+

然后，您可以应用groupBy聚合。