Как удалить столбцы на основе нескольких фильтров в кадре данных, используя PySpark?

Question

У меня есть список допустимых значений, которые может иметь ячейка.Если одна ячейка в столбце неверна, мне нужно удалить весь столбец.Я понимаю, что есть ответы на удаление строк в определенном столбце, но здесь я вместо этого удаляю весь столбец, даже если одна ячейка в нем недействительна.Условия для действительного / недействительного состоят в том, что ячейка может иметь только три значения: ['Messi', 'Ronaldo', 'Virgil']

Я попытался прочитать о фильтрации, но все, что я мог видеть, это фильтрация столбцов и удаление строк.Например, в этом вопросе.Я также читал, что следует избегать слишком большого сканирования и перестановки в Spark, с чем я согласен.

Я смотрю не только на решение для кода, но и на готовый код, предоставляемый PySpark.Я надеюсь, что это не выходит за рамки SO-ответа.

Для следующего входного фрейма данных:

| Column 1      | Column 2      | Column 3      | Column 4      | Column 5      |
| --------------| --------------| --------------| --------------| --------------|
|  Ronaldo      | Salah         |  Messi        |               |Salah          |
|  Ronaldo      | Messi         |  Virgil       |  Messi        | null          |
|  Ronaldo      | Ronaldo       |  Messi        |  Ronaldo      | null          |

Ожидается следующий вывод:

| Column 1      | Column 2      |
| --------------| --------------| 
|  Ronaldo      | Messi         |
|  Ronaldo      | Virgil        |
|  Ronaldo      | Messi         |

Answer 1

Я не только смотрю на решение для кода, но и на готовый код, предоставляемый PySpark.

К сожалению, Spark предназначен для параллельной работы в строкена основе строки.Фильтрация столбцов - это не то, для чего будет решение «готового кода».

Тем не менее, вот один из подходов, который вы можете использовать:

Сначала соберите счетчикинедопустимые элементы в каждом столбце.

from pyspark.sql.functions import col, lit, sum as _sum, when

valid = ['Messi', 'Ronaldo', 'Virgil']
invalid_counts = df.select(
    *[_sum(when(col(c).isin(valid), lit(0)).otherwise(lit(1))).alias(c) for c in df.columns]
).collect()
print(invalid_counts)
#[Row(Column 1=0, Column 2=1, Column 3=0, Column 4=1, Column 5=3)]

Этот вывод будет списком только с одним элементом.Вы можете перебирать элементы в этом элементе, чтобы найти столбцы для хранения.

valid_columns = [k for k,v in invalid_counts[0].asDict().items() if v == 0]
print(valid_columns)
#['Column 3', 'Column 1']

Теперь просто выберите эти столбцы из исходного DataFrame.Сначала вы можете отсортировать valid_columns, используя list.index, если хотите сохранить исходный порядок столбцов.

valid_columns = sorted(valid_columns, key=df.columns.index)
df.select(valid_columns).show()
#+--------+--------+
#|Column 1|Column 3|
#+--------+--------+
#| Ronaldo|   Messi|
#| Ronaldo|  Virgil|
#| Ronaldo|   Messi|
#+--------+--------+