Суммирование значений по каждой строке как логическое (PySpark)

Question

В настоящее время у меня есть фрейм данных PySpark, в котором много столбцов, заполненных целочисленными значениями. Многие из этих столбцов имеют количество нулей. Я хотел бы найти способ суммировать, сколько столбцов имеет число больше нуля .

Другими словами, я хотел бы использовать подход, который суммирует значения по строке, где все столбцы для данной строки фактически являются логическими (хотя преобразование типа данных может не потребоваться). Несколько столбцов в моей таблице представляют собой дату и время или строку, поэтому в идеале у меня должен быть подход, который сначала выбирает цифры c столбцы.

Пример текущего кадра данных и требуемый вывод

+---+---------- +----------+------------            
|USER|   DATE   |COUNT_COL1| COUNT_COL2|...     DESIRED COLUMN
+---+---------- +----------+------------ 
| b | 7/1/2019 |  12      |     1     |              2        (2 columns are non-zero)
| a | 6/9/2019 |  0       |     5     |              1
| c | 1/1/2019 |  0       |     0     |              0

Pandas: В качестве примера, в pandas это может быть достигнуто путем выбора числовых столбцов c, преобразования в bool и суммирования с осью = 1. Я ищу эквивалент PySpark.

test_cols=list(pandas_df.select_dtypes(include=[np.number]).columns.values)
pandas_df[test_cols].astype(bool).sum(axis=1)

Answer 1

Для чисел вы можете сделать это, создав array всех столбцов с integer values (используя df.dtypes), а затем используйте higher order functions. В этом случае я использовал filter, чтобы избавиться от всех 0, а затем использовал размер, чтобы получить количество всех ненулевых элементов в строке. (spark2.4+)

from pyspark.sql import functions as F
df.withColumn("arr", F.array(*[F.col(i[0]) for i in df.dtypes if i[1] in ['int','bigint']]))\
  .withColumn("DESIRED COLUMN", F.expr("""size(filter(arr,x->x!=0))""")).drop("arr").show()

#+----+--------+----------+----------+--------------+
#|USER|    DATE|COUNT_COL1|COUNT_COL2|DESIRED COLUMN|
#+----+--------+----------+----------+--------------+
#|   b|7/1/2019|        12|         1|             2|
#|   a|6/9/2019|         0|         5|             1|
#|   c|1/1/2019|         0|         0|             0|
#+----+--------+----------+----------+--------------+

Answer 2

Допустим, у вас ниже df:

df.show()
df.printSchema()

+---+---+---+---+
|_c0|_c1|_c2|_c3|
+---+---+---+---+
|  a|  1|  2|  3|
|  a|  0|  2|  1|
|  a|  0|  0|  1|
|  a|  0|  0|  0|
+---+---+---+---+

root
 |-- _c0: string (nullable = true)
 |-- _c1: string (nullable = true)
 |-- _c2: string (nullable = true)
 |-- _c3: string (nullable = true)

Используя оператор case когда вы можете проверить, является ли столбец цифрой c, а затем, если он больше 0. На следующем шаге f.size будет счетчик возвратов благодаря f.array_remove, в котором остались только столбцы со значением True.

from pyspark.sql import functions as f
cols = [f.when(f.length(f.regexp_replace(f.col(x), '\\d+', '')) > 0, False).otherwise(f.col(x).cast('int') > 0) for x in df2.columns]
df.select("*", f.size(f.array_remove(f.array(*cols), False)).alias("count")).show()

+---+---+---+---+-----+
|_c0|_c1|_c2|_c3|count|
+---+---+---+---+-----+
|  a|  1|  2|  3|    3|
|  a|  0|  2|  1|    2|
|  a|  0|  0|  1|    1|
|  a|  0|  0|  0|    0|
+---+---+---+---+-----+