Удалить пустые столбцы из фрейма данных pyspark

Question

У меня очень грязный csv, в котором есть несколько столбцов с только нулевыми значениями.

Я хотел бы удалить их. Я пытаюсь выбрать все столбцы, в которых число пустых значений в столбце не равно количеству строк.

clean_df = bucketed_df.select([c for c in bucketed_df.columns if count(when(isnull(c), c)) not bucketed_df.count()])

Однако я получаю эту ошибку:

SyntaxError: invalid syntax
  File "<command-2213215314329625>", line 1
    clean_df = bucketed_df.select([c for c in bucketed_df.columns if count(when(isnull(c), c)) not bucketed_df.count()])
                                                                                                             ^
SyntaxError: invalid syntax

Если бы кто-нибудь мог помочь мне избавиться от этих грязных колонн, это было бы здорово.

Answer 1

Может использоваться функция "min", а столбцы с нулевыми значениями могут быть удалены. Здесь на Scala, можно легко перевести на Python:

// data sample
val df = Seq(("Bug", null.asInstanceOf[Integer], null.asInstanceOf[String]),
  ("Termit", null.asInstanceOf[Integer], null.asInstanceOf[String]))
  .toDF("name", "size", "type")
val fieldNames = df.schema.fieldNames

// get null values 
val fieldExpressions = fieldNames.map(c => min(c).alias(c))
val firstRow = df.select(fieldExpressions: _*).collect().head

val fieldsToDrop = fieldNames.filter(f => firstRow.isNullAt(fieldNames.indexOf(f)))

Answer 2

У меня очень мало опыта работы с pyspark, но, возможно, неплохо было бы создать фрейм данных с счетчиками и преобразовать его в pandas, поскольку у фрейма count будет одна строка:

Начиная с фрейм данных, который выглядит как показано ниже и сохраняется как null_df

+---+---+---+----+
|  A|  B|  C|   D|
+---+---+---+----+
|  1|  a|  b|null|
|  2|  c|  d|null|
|  3|  e|  f|null|
+---+---+---+----+

---

import pyspark.sql.functions as F

counts = null_df.select([F.count(i).alias(i) for i in null_df.columns]).toPandas()
output = null_df.select(*counts.columns[counts.ne(0).iloc[0]])

Или даже преобразование всей первой строки в словарь, а затем l oop по словарю

counts1 = null_df.select([F.count(i).alias(i) for i in null_df.columns])
output2 = null_df.select([k for k,v in counts1.first().asDict().items() if v >0])

, что дает следующее:

+---+---+---+
|  A|  B|  C|
+---+---+---+
|  1|  a|  b|
|  2|  c|  d|
|  3|  e|  f|
+---+---+---+

тесты, как проверено в моей системе:

%%timeit
counts = null_df.select([F.count(i).alias(i) for i in null_df.columns]).toPandas()
output = null_df.select(*counts.columns[counts.ne(0).iloc[0]])
#8.73 s ± 412 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%%timeit
counts1 = null_df.select([F.count(i).alias(i) for i in null_df.columns])
output2 = null_df.select([k for k,v in counts1.first().asDict().items() if v >0])
#9.43 s ± 253 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%%timeit
counts1 = null_df.select([F.count(i).alias(i) for i in null_df.columns])
output1 = null_df.select([c for c in counts1.columns if counts1[[c]].first()[c] > 0])
#35.3 s ± 1 s per loop (mean ± std. dev. of 7 runs, 1 loop each)

Answer 3

Есть 2 способа решения этой проблемы,

1) Напишите функцию UDF для добавления столбца, в котором значение столбца равно 1, если значение требуемого столбца (столбца, который вы проверяете на NULL) NULL, затем возьмите сумму столбца, если сумма равна количеству строк, затем отбросьте столбец

2) Используйте библиотеку amazon spark dq (библиотека качества данных с открытым исходным кодом для spark), библиотека имеет функции для профилирования ваших данных, один из столбцов, который возвращает dq, является коэффициентом полноты для каждого столбца. Если коэффициент полноты равен 1, тогда все значения столбца равны NULL, вы можете удалить этот столбец, лично я считаю, что эта опция хороша, потому что он предназначен для проверки качества данных с использованием spark

. В приведенных ниже ссылках приведено множество примеров

https://aws.amazon.com/blogs/big-data/test-data-quality-at-scale-with-deequ/ https://github.com/awslabs/deequ