агрегировать столбец на основе количества значений других столбцов в фрейме данных pyspark

Question

Я хотел бы выполнить несколько агрегатов для таблицы куста pyspark.

моя таблица:

  id   value_tier ($)
  105   5
  117   5
  108   10
  110   12
  105   10
  112   10

Мне нужно получить количество идентификаторов, которые появляются только в одном "value_tier" .

value_tier   num
   5          1 -- for 117
   10         2 -- for 108 and 112
   12         1 -- for 110

Здесь 105 не считается, потому что оно встречается в двух value_tiers.

    5 and 10

My sql DDL работает, но долго и некрасиво. Хотелось бы еще одного элегантного. спасибо

Answer 1

В DataFrameAPI используйте groupBy и agg с функцией collect_list.

df1.show()
#+---+----------+
#| id|value_tier|
#+---+----------+
#|105|         5|
#|117|         5|
#|108|        10|
#|110|        12|
#|105|        10|
#|112|        10|
#+---+----------+

from pyspark.sql.functions import *

df1.groupBy("id").
agg(concat_ws(',',collect_list(col("value_tier"))).alias("value_tier")).\
filter(size(split(col("value_tier"),",")) <=1).\
groupBy("value_tier").\
agg(count(col("id")).alias("num"),concat_ws(",",collect_list(col("id"))).alias("ids")).\
show()
#+----------+---+-------+
#|value_tier|num|    ids|
#+----------+---+-------+
#|         5|  1|    117|
#|        10|  2|112,108|
#|        12|  1|    110|
#+----------+---+-------+

#use collect_set to eliminate duplicates
df1.groupBy("id").
agg(concat_ws(',',collect_set(col("value_tier"))).alias("value_tier")).\
filter(size(split(col("value_tier"),",")) <=1).\
groupBy("value_tier").\
agg(count(col("id")).alias("num"),concat_ws(",",collect_list(col("id"))).alias("ids")).\
show()

Answer 2

В SQL можно использовать not exists и агрегирование:

sélect value_tier, count(*) cnt
from mytable t
where not exists(
    select 1
    from mytable t1
    where t1.value_tier = t.value_tier and t1.id <> t.id
)
group by value_tier