将值透视到PySpark数据框架中的现有列

Question

我在pyspark中有一个如下的数据帧。你知道吗

+---+-------------+----+
| id|       device| val|
+---+-------------+----+
|  3|      mac pro|   1|          
|  1|       iphone|   2|
|  1|android phone|   2|
|  1|   windows pc|   2|
|  1|   spy camera|   2|
|  2|   spy camera|   3|
|  2|       iphone|   3|
|  3|   spy camera|   1|
|  3|         cctv|   1|
+---+-------------+----+

我想根据下面的lists填充一些列

phone_list = ['iphone', 'android phone', 'nokia']
pc_list = ['windows pc', 'mac pro']
security_list = ['spy camera']
ucg_list = ['ipad']

我做了如下的事情。你知道吗

from pyspark.sql.functions import col, when, lit 
from pyspark.sql.types import IntegerType
df1 = df.withColumn('phones', lit(None).cast(IntegerType())).withColumn('pc', lit(None).cast(IntegerType())).withColumn('security', lit(None).cast(IntegerType())).withColumn('null', lit(None).cast(IntegerType())).withColumn('ucg', lit(None).cast(IntegerType()))

import pyspark.sql.functions as F

df1.withColumn('cat', 
    F.when(df1.device.isin(phone_list), 'phones').otherwise(
    F.when(df1.device.isin(pc_list), 'pc').otherwise(
    F.when(df1.device.isin(security_list), 'security')))
).groupBy('id', 'phones', 'pc', 'security', 'null', 'ucg').pivot('cat').agg(F.count('cat')).show()

我正在接收的输出

+---+------+----+--------+----+----+----+---+------+--------+
| id|phones|  pc|security|null| ucg|null| pc|phones|security|
+---+------+----+--------+----+----+----+---+------+--------+
|  3|  null|null|    null|null|null|   0|  1|     0|       1|
|  2|  null|null|    null|null|null|   0|  0|     1|       1|
|  1|  null|null|    null|null|null|   0|  1|     2|       1|
+---+------+----+--------+----+----+----+---+------+--------+

我想要的是首先基于列表名创建列，然后填充值

预期产量

+---+------+---+------+--------+----+
| id|   ucg| pc|phones|security|null|
+---+------+---+------+--------+----+
|  1|     0|  1|     2|       1|   0|
|  2|     0|  0|     1|       1|   0|
|  3|     0|  1|     0|       1|   1|
+---+------+---+------+--------+----+

我怎样才能得到我想要的？你知道吗

编辑

当我做下面这些的时候

df1 = df.withColumn('cat', 
    f.when(df.device.isin(phone_list), 'phones').otherwise(
    f.when(df.device.isin(pc_list), 'pc').otherwise(
    f.when(df.device.isin(ucg_list), 'ucg').otherwise(
    f.when(df.device.isin(security_list), 'security')))))

输出为

+---+-------------+---+--------+
| id|       device|val|     cat|
+---+-------------+---+--------+
|  3|      mac pro|  1|      pc|
|  3|   spy camera|  1|security|
|  3|         cctv|  1|    null|
|  1|       iphone|  2|  phones|
|  1|android phone|  2|  phones|
|  1|   windows pc|  2|      pc|
|  1|   spy camera|  2|security|
|  2|   spy camera|  3|security|
|  2|       iphone|  3|  phones|
+---+-------------+---+--------+

在输出中可以看到id3在cat列中有一个null值

Answer 1

仅仅为groupBy创建None并填充'phones', 'pc', 'ucg', 'security', 'null'列是没有意义的。带id的分组和上面所有带null或只按id分组的列都是相同的。你知道吗

您可以改为查找实际数据透视列和预期列之间的差异，然后创建并填充0

所以下面的内容应该对你有用

phone_list = ['iphone', 'android phone', 'nokia']
pc_list = ['windows pc', 'mac pro']
security_list = ['spy camera']
ucg_list = ['ipad']

from pyspark.sql import functions as f
df = df.withColumn('cat',
               f.when(df.device.isin(phone_list), 'phones').otherwise(
                 f.when(df.device.isin(pc_list), 'pc').otherwise(
                   f.when(df.device.isin(ucg_list), 'ucg').otherwise(
                     f.when(df.device.isin(security_list), 'security'))))
               )\
    .groupBy('id').pivot('cat').agg(f.count('val'))\
    .na.fill(0)\

columnList = ['phones', 'pc', 'ucg', 'security', 'null']
actualcolumnList = df.columns[1:]

diffColumns = [x for x in columnList if x not in actualcolumnList]

for y in diffColumns:
    df = df.withColumn(y, f.lit(0))

df.show(truncate=False)

这应该给你

+ -+  + -+   +    + -+
|id |null|pc |phones|security|ucg|
+ -+  + -+   +    + -+
|3  |1   |1  |0     |1       |0  |
|1  |0   |1  |2     |1       |0  |
|2  |0   |0  |1     |1       |0  |
+ -+  + -+   +    + -+

我希望答案是有帮助的