Есть ли способ в pyspark для подсчета уникальных значений

Question

У меня есть искровой фрейм данных (12 м x 132), и я пытаюсь рассчитать количество уникальных значений по столбцам и удалить столбцы, которые имеют только 1 уникальное значение.

До сих пор я использовал nunique pandas nunique следующим образом:

import pandas as pd

df = sql_dw.read_table(<table>)
df_p = df.toPandas()

nun = df_p.nunique(axis=0)
nundf = pd.DataFrame({'atr':nun.index, 'countU':nun.values})

dropped = []
for i, j in nundf.values:
  if j == 1:
    dropped.append(i)
    df = df.drop(i)
print(dropped)

Есть ли способ сделать это более естественным для искры - т.е. не использовать панд?

Answer 1

Приношу свои извинения, поскольку у меня нет решения в pyspark, но в чистом искре, которое может быть передано или использовано в случае, если вы не можете найти способ pyspark. затем, используя foreach, проверьте, какие столбцы имеют различное число 1, затем добавьте их в пустой список.

Оттуда вы можете использовать список в качестве фильтра и отбрасывать эти столбцы с вашего фрейма данных.

    var list_of_columns: List[String] = ()

    df_p.columns.foreach{c => 
      if (df_p.select(c).distinct.count == 1)
         list_of_columns ++= List(c)

    df_p_new = df_p.drop(list_of_columns:_*)

Answer 2

Вы можете сгруппировать свой df по этому столбцу и посчитать отличное значение этого столбца:

df = df.groupBy("column_name").agg(countDistinct("column_name").alias("distinct_count"))

И затем отфильтровать свой df по строке, которая имеет более 1 однозначных_числов:

df = df.filter(df.distinct_count > 1)

Answer 3

Пожалуйста, посмотрите на прокомментированный пример ниже. Решение требует больше знаний о Python как специфических для pyspark.

import pyspark.sql.functions as F
#creating a dataframe
columns = ['asin' ,'ctx' ,'fo' ]

l = [('ASIN1','CTX1','FO1')
,('ASIN1','CTX1','FO1')
,('ASIN1','CTX1','FO2')
,('ASIN1','CTX2','FO1')
,('ASIN1','CTX2','FO2')
,('ASIN1','CTX2','FO2')
,('ASIN1','CTX2','FO3')
,('ASIN1','CTX3','FO1')
,('ASIN1','CTX3','FO3')]

df=spark.createDataFrame(l, columns)

df.show()
#we create a list of functions we want to apply
#in this case countDistinct for each column
expr = [F.countDistinct(c).alias(c) for c in df.columns]

#we apply those functions
countdf =  df.select(*expr)
#this df has just one row
countdf.show()

#we extract the columns which have just one value
cols2drop = [k for k,v in countdf.collect()[0].asDict().items() if v == 1]
df.drop(*cols2drop).show()

Вывод:

+-----+----+---+
| asin| ctx| fo|
+-----+----+---+
|ASIN1|CTX1|FO1|
|ASIN1|CTX1|FO1|
|ASIN1|CTX1|FO2|
|ASIN1|CTX2|FO1|
|ASIN1|CTX2|FO2|
|ASIN1|CTX2|FO2|
|ASIN1|CTX2|FO3|
|ASIN1|CTX3|FO1|
|ASIN1|CTX3|FO3|
+-----+----+---+

+----+---+---+
|asin|ctx| fo|
+----+---+---+
|   1|  3|  3|
+----+---+---+

+----+---+
| ctx| fo|
+----+---+
|CTX1|FO1|
|CTX1|FO1|
|CTX1|FO2|
|CTX2|FO1|
|CTX2|FO2|
|CTX2|FO2|
|CTX2|FO3|
|CTX3|FO1|
|CTX3|FO3|
+----+---+