如何根据pyspark中的行和列条件过滤多行

Question

我想根据“值”列过滤多行。例如，我想从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)))) 

Answer 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 时，只有其中一个子句为真，才能将行包含在结果数据框中。