从 PySpark 数据框中选择“独占行”

Question

我有一个这样的 PySpark 数据框：

+----------+-----+
|account_no|types|
+----------+-----+
|         1|    K|
|         1|    A|
|         1|    S|
|         2|    M|
|         2|    D|
|         2|    S|
|         3|    S|
|         3|    S|
|         4|    M|
|         5|    K|
|         1|    S|
|         6|    S|
+----------+-----+

我正在尝试选择“S”存在的帐号。 例如：即使'1'的类型='S'，我也不会选择它，因为它还有其他类型。但我会选择 3 和 6，因为它们只有一种类型“S”。

我现在正在做的是： - 首先获取所有存在“K”的帐户并将其删除；在此示例中删除了“1”和“5” - 其次查找存在“D”的所有帐户并删除它们，从而删除“2” - 第三次查找所有存在“M”的帐户，并删除“4”（“2”也有“M”，但在第 2 步被删除） - 第四找到所有存在“A”的账户，并删除它们

所以，现在“1”、“2”、“4”和“5”被删除，我得到“3”和“6”，它们具有唯一的“S”。

但这是一个漫长的过程，我该如何优化呢？ 谢谢

Answer 1

另一种选择是在窗口上计数不同，然后 filter 其中不同 count == 1 和 types == S ，对于排序，您可以分配一个单调递增的 id，然后 orderBy 相同。

from pyspark.sql import functions as F
W = Window.partitionBy('account_no')

out = (df.withColumn("idx",F.monotonically_increasing_id())
   .withColumn("Distinct",F.approx_count_distinct(F.col("types")).over(W)).orderBy("idx")
   .filter("Distinct==1 AND types =='S'")).drop('idx','Distinct')

---

out.show()

+----------+-----+
|account_no|types|
+----------+-----+
|         3|    S|
|         3|    S|
|         6|    S|
+----------+-----+

Answer 2

一种方法是使用 Window 函数。首先，我们在每个 account_no 分组中获得 S 数量的 sum。然后我们将其与 group 的条目总数进行比较，在过滤器中，如果 它们匹配，我们 @987654326 @。

from pyspark.sql import functions as F
from pyspark.sql.window import Window

w=Window().partitionBy("account_no")
w1=Window().partitionBy("account_no").orderBy("types")

df.withColumn("sum_S", F.sum(F.when(F.col("types")=='S', F.lit(1)).otherwise(F.lit(0))).over(w))\
  .withColumn("total", F.max(F.row_number().over(w1)).over(w))\
  .filter('total=sum_S').drop("total","Sum_S").show()

#+----------+-----+
#|account_no|types|
#+----------+-----+
#|         6|    S|
#|         3|    S|
#|         3|    S|
#+----------+-----+

Answer 3

您可以简单地检测一个帐户具有的不同类型的数量，然后过滤只有 1 个不同类型的“S”帐户。

这是我的代码：

from pyspark.sql.functions import countDistinct

data = [(1, 'k'),
        (1, 'a'),
        (1, 's'),
        (2, 'm'),
        (2, 'd'),
        (2, 's'),
        (3, 's'),
        (3, 's'),
        (4, 'm'),
        (5, 'k'),
        (1, 's'),
        (6, 's')]

df = spark.createDataFrame(data, ['account_no', 'types']).distinct()

exclusive_s_accounts = (df.groupBy('account_no').agg(countDistinct('types').alias('distinct_count'))
                        .join(df, 'account_no')
                        .where((col('types') == 's') & (col('distinct_count') == 1))
                        .drop('distinct_count'))

Answer 4

另一种替代方法是获取一列下的所有类型，然后应用过滤操作排除具有非“S”值的类型。

from pyspark.sql.functions import concat_ws
from pyspark.sql.functions import collectivist
from pyspark.sql.functions import col
df = spark.read.csv("/Users/Downloads/account.csv", header=True, inferSchema=True, sep=",")
type_df = df.groupBy("account_no").agg(concat_ws(",",     collect_list("types")).alias("all_types")).select(col("account_no"),     col("all_types"))

+----------+---------+
|account_no|all_types|
+----------+---------+
|         1|  K,A,S,S|
|         6|        S|
|         3|      S,S|
|         5|        K|
|         4|        M|
|         2|    M,D,S|
+----------+---------+

further filtering using regular expression
only_s_df =  type_df.withColumn("S_status",F.col("all_types").rlike("K|A|M|D"))
only_s_df.show()
+----------+---------+----------+
|account_no|all_types|S_status  |
+----------+---------+----------+
|         1|  K,A,S,S|      true|
|         6|        S|     false|
|         3|      S,S|     false|
|         5|        K|      true|
|         4|        M|      true|
|         2|    M,D,S|      true|
+----------+---------+----------+

希望通过这种方式可以得到答案并进一步处理。