如何根据pyspark中的行和列条件过滤多行
Posted
技术标签:
【中文标题】如何根据pyspark中的行和列条件过滤多行【英文标题】:How to filter multiple rows based on rows and columns condition in pyspark 【发布时间】:2022-01-16 22:55:29 【问题描述】:我想根据“值”列过滤多行。例如,我想从channel_name
列中过滤velocity
value>=1 & value <=5
,我想从channel_name
列中过滤Temp
value>=0 & value <=2
。下面是我的 Pysaprk DF。
start_timestamp | channel_name | value |
---|---|---|
2020-11-02 08:51:50 | velocity | 1 |
2020-11-02 09:14:29 | Temp | 0 |
2020-11-02 09:18:32 | velocity | 0 |
2020-11-02 09:32:42 | velocity | 4 |
2020-11-03 13:06:03 | Temp | 2 |
2020-11-03 13:10:01 | Temp | 1 |
2020-11-03 13:54:38 | Temp | 5 |
2020-11-03 14:46:25 | velocity | 5 |
2020-11-03 14:57:31 | Kilometer | 6 |
2020-11-03 15:07:07 | Kilometer | 7 |
预期 DF:
start_timestamp | channel_name | value |
---|---|---|
2020-11-02 08:51:50 | velocity | 1 |
2020-11-02 09:32:42 | velocity | 4 |
2020-11-03 14:46:25 | velocity | 5 |
2020-11-02 09:14:29 | Temp | 0 |
2020-11-03 13:06:03 | Temp | 2 |
2020-11-03 13:10:01 | Temp | 1 |
我尝试了 channel_name Velocity
,它工作正常。
df1=df.filter((df.channel_name == "velocity") & (df.interpreted_value >= 1 ) & (df.interpreted_value <= 5))
但我不知道如何为多个channel_name
(如Velocity
和Temp
)做到这一点:下面是代码,也让我知道这是否是正确的做法或我该怎么做。
df1=df.filter(((df.channel_name == "velocity") & (df.interpreted_value >= 1 ) &
(df.interpreted_value <= 5))) &
((df.channel_name == "Temp") & (df.interpreted_value >= 0 ) &
(df.interpreted_value <= 2))))
【问题讨论】:
【参考方案1】:在组合子句时,您需要使用 or (|
) 而不是 and(&
) 运算符:
import pyspark.sql.functions as F
import pyspark.sql.types as T
df = spark.createDataFrame([
("2020-11-02 08:51:50", "velocity", 1),
("2020-11-02 09:14:29", "Temp", 0),
("2020-11-02 09:18:32", "velocity", 0),
("2020-11-02 09:32:42", "velocity", 4),
("2020-11-03 13:06:03", "Temp", 2),
("2020-11-03 13:10:01", "Temp", 1),
("2020-11-03 13:54:38", "Temp", 5),
("2020-11-03 14:46:25", "velocity", 5),
("2020-11-03 14:57:31", "Kilometer",6),
("2020-11-03 15:07:07", "Kilometer", 7)],
["start_timestamp", "channel_name", "value"]).withColumn("start_timestamp", F.to_timestamp("start_timestamp"))
df_filtered = df.filter((((df.channel_name == "velocity") & (df.value >= 1 ) &
(df.value <= 5))) | # or instead of and
((df.channel_name == "Temp") & (df.value >= 0 ) &
(df.value <= 2)))
df_filtered.show()
输出:
+-------------------+------------+-----+
| start_timestamp|channel_name|value|
+-------------------+------------+-----+
|2020-11-02 08:51:50| velocity| 1|
|2020-11-02 09:14:29| Temp| 0|
|2020-11-02 09:32:42| velocity| 4|
|2020-11-03 13:06:03| Temp| 2|
|2020-11-03 13:10:01| Temp| 1|
|2020-11-03 14:46:25| velocity| 5|
+-------------------+------------+-----+
您当前应用的过滤器不会返回任何内容,因为您首先检查频道名称是否等于一个特定字符串,然后检查它是否等于另一个特定字符串。使用 or 时,只有其中一个子句为真,才能将行包含在结果数据框中。
【讨论】:
谢谢,效果很好。仍然在逻辑上感到困惑 AND 应该正确插入 OR 吗?因为我希望两个括号条件都为真。 您希望保留任一方括号返回 true 的每一行。你永远不能让两个括号都返回 true,因为两个字符串比较永远不会同时为 true。以上是关于如何根据pyspark中的行和列条件过滤多行的主要内容,如果未能解决你的问题,请参考以下文章
如何转换由 | 分隔的顺序数据并且在 pyspark 中的行和列中没有换行符