Заполните отдельный столбец на основе другого столбца в PySpark

Question

У меня есть фрейм данных, как показано ниже в PySpark.Я хочу выбрать serial_num, devicetype, device_model и distinct of timestamp for each serial_num из приведенного ниже фрейма данных:

+-------------+-----------------+---------------+------------------------+
| serial_num  |   devicetype    | device_model  |        timestamp       |
+-------------+-----------------+---------------+------------------------+
| 58172A0396  |                 |               | 2003-01-02 17:37:15.0  |
| 58172A0396  |                 |               | 2003-01-02 17:37:15.0  |
| 46C5Y00693  | Mac Pro         | Mac PC        | 2018-01-03 17:17:23.0  |
| 1737K7008F  | Windows PC      | Windows PC    | 2018-01-05 11:12:31.0  |
| 1737K7008F  | Network Device  | Unknown       | 2018-01-05 11:12:31.0  |
| 1737K7008F  | Network Device  | Unknown       | 2018-01-05 11:12:31.0  |
| 1737K7008F  | Network Device  |               | 2018-01-06 03:12:52.0  |
| 1737K7008F  | Windows PC      | Windows PC    | 2018-01-06 03:12:52.0  |
| 1737K7008F  | Network Device  | Unknown       | 2018-01-06 03:12:52.0  |
| 1665NF01F3  | Network Device  | Unknown       | 2018-01-07 03:42:34.0  |
+----------------+-----------------+---------------+---------------------+

Я пробовал, как показано ниже

df1 = df.select('serial_num', 'devicetype', 'device_model', f.count('distinct timestamp').over(Window.partitionBy('serial_num')).alias('val')

Результат Iхочу это:

+-------------+-----------------+---------------+-----+
| serial_num  |   devicetype    | device_model  |count|
+-------------+-----------------+---------------+-----+
| 58172A0396  |                 |               |  1  |
| 58172A0396  |                 |               |  1  |
| 46C5Y00693  | Mac Pro         | Mac PC        |  1  |
| 1737K7008F  | Windows PC      | Windows PC    |  2  |
| 1737K7008F  | Network Device  | Unknown       |  2  |
| 1737K7008F  | Network Device  | Unknown       |  2  |
| 1737K7008F  | Network Device  |               |  2  |
| 1737K7008F  | Windows PC      | Windows PC    |  2  |
| 1737K7008F  | Network Device  | Unknown       |  2  |
| 1665NF01F3  | Network Device  | Unknown       |  1  |
+-------------+-----------------+---------------+-----+

Как мне этого добиться?

Answer 1

Простой groupBy и считать будет работать.

val data=Array(("58172A0396","","","2003-01-02 17:37:15.0"),
("58172A0396","","","2003-01-02 17:37:15.0"),
("46C5Y00693"," Mac Pro","Mac PC","2018-01-03 17:17:23.0"),
("1737K7008F"," Windows PC","Windows PC","2018-01-05 11:12:31.0"),
("1737K7008F"," Network Device","Unknown","2018-01-05 11:12:31.0"),
("1737K7008F"," Network Device","Unknown","2018-01-05 11:12:31.0"),
("1737K7008F"," Network Device","","2018-01-06 03:12:52.0"),
("1737K7008F"," Windows PC","Windows PC","2018-01-06 03:12:52.0"),
("1737K7008F"," Network Device","Unknown","2018-01-06 03:12:52.0"),
("1665NF01F3"," Network Device","Unknown","2018-01-07 03:42:34.0"))

val rdd = sc.parallelize(data)

val df = rdd.toDF("serial_num","devicetype","device_model","timestamp")

val df1 = df.groupBy("timestamp","serial_num","devicetype","device_model").count

Answer 2

К сожалению, countDistinct не поддерживается для Windows.Однако комбинация collect_set и size может быть использована для достижения того же конечного результата.Это поддерживается только в Spark 2.0 + версиях, используйте следующее:

import pyspark.sql.funcions as F

w = Window.partitionBy('serial_num')
df1 = df.select(..., F.size(F.collect_set('timestamp').over(w)).alias('count'))

---

Для старых версий Spark , что вы можете сделать, это использовать groupby и countDistinct для создания нового кадра данных со всеми значениями.Затем join этот кадр данных вместе с исходным.

df2 = df.groupby('serial_num').agg(F.countDistinct('timestamp').alias('count'))
df1 = df.join(df2, 'serial_num')