拆分 pyspark 数据框列的内容并将它们聚合到新列中

Question

我正在尝试提取和拆分 pyspark 数据框列中的数据，然后将其聚合到一个新列中。

输入表。

+--+-----------+
|id|description|
+--+-----------+
|1 |  3:2,3|2:1|
|2 |  2        |
|3 |  2:12,16  |
|4 |  3:2,4,6  |
|5 |  2        |
|6 |  2:3,7|2:3|
+--------------+

所需的输出。

+--+-----------+-------+-----------+
|id|description|sum_emp|org_changed|
+--+-----------+-------+-----------+
|1 |  3:2,3|2:1| 5     | 3         |
|2 |  2        | 2     | 0         |        
|3 |  2:12,16  | 2     | 2         |
|4 |  3:2,4,6  | 3     | 3         |
|5 |  2        | 2     | 0         |
|6 |  2:3,7|2:3| 4     | 3         |
+--------------+-------+-----------+

在“:”之前，应该添加值。 “：”后的值将被计算在内。该|标记记录中的移位（可以忽略）

有些数据点长达 2:3,4,5|3:4,6,3|4:3,7,8
任何帮助将不胜感激

场景解释：

以第 6 个 id 为例。 6 指的是商业单位 ID。 “描述”列描述了该给定单元内的团队。

现在对于值 2:3,7|2:3 的含义如下： 1）拳头 2 后跟 3&7 = 团队中有 2 个人，其中一个已经在另一个组织工作了 3 年和 7 年（也许是第二个人第一个公司） 2）第二个 2 后跟 3 = 一个子团队又有 2 个人，并且 1 个人在另一个组织工作了 3 年。

期望的输出： sum_emp = 给定 biz 单位的员工总数。 org_changed = 该业务部门中已更改的组织人员总数。

Answer 1

首先让我们创建我们的数据框：

df = spark.createDataFrame(
    sc.parallelize([[1,"3:2,3|2:1"],
                    [2,"2"],
                    [3,"2:12,16"],
                    [4,"3:2,4,6"],
                    [5,"2"],
                    [6,"2:3,7|2:3"]]), 
    ["id","description"])

    +---+-----------+
    | id|description|
    +---+-----------+
    |  1|  3:2,3|2:1|
    |  2|          2|
    |  3|    2:12,16|
    |  4|    3:2,4,6|
    |  5|          2|
    |  6|  2:3,7|2:3|
    +---+-----------+

首先，我们将拆分记录并分解生成的数组，这样我们每行只有一条记录：

import pyspark.sql.functions as psf

df = df.withColumn(
    "record", 
    psf.explode(psf.split("description", '\|'))
)

    +---+-----------+-------+
    | id|description| record|
    +---+-----------+-------+
    |  1|  3:2,3|2:1|  3:2,3|
    |  1|  3:2,3|2:1|    2:1|
    |  2|          2|      2|
    |  3|    2:12,16|2:12,16|
    |  4|    3:2,4,6|3:2,4,6|
    |  5|          2|      2|
    |  6|  2:3,7|2:3|  2:3,7|
    |  6|  2:3,7|2:3|    2:3|
    +---+-----------+-------+

现在我们将记录分为玩家数量和年份列表：

df = df.withColumn(
    "record", 
    psf.split("record", ':')
).withColumn(
    "nb_players", 
    psf.col("record")[0]
).withColumn(
    "years", 
    psf.split(psf.col("record")[1], ',')
)

    +---+-----------+----------+----------+---------+
    | id|description|    record|nb_players|    years|
    +---+-----------+----------+----------+---------+
    |  1|  3:2,3|2:1|  [3, 2,3]|         3|   [2, 3]|
    |  1|  3:2,3|2:1|    [2, 1]|         2|      [1]|
    |  2|          2|       [2]|         2|     null|
    |  3|    2:12,16|[2, 12,16]|         2| [12, 16]|
    |  4|    3:2,4,6|[3, 2,4,6]|         3|[2, 4, 6]|
    |  5|          2|       [2]|         2|     null|
    |  6|  2:3,7|2:3|  [2, 3,7]|         2|   [3, 7]|
    |  6|  2:3,7|2:3|    [2, 3]|         2|      [3]|
    +---+-----------+----------+----------+---------+

最后，我们要对每个id 的玩家数量和years 的长度求和：

df = df.withColumn(
    "years_size", 
    psf.when(psf.size("years") > 0, psf.size("years")).otherwise(0)
).groupby("id").agg(
    psf.first("description").alias("description"), 
    psf.sum("nb_players").alias("sum_emp"), 
    psf.sum("years_size").alias("org_changed")
).sort("id").show()

    +---+-----------+-------+-----------+
    | id|description|sum_emp|org_changed|
    +---+-----------+-------+-----------+
    |  1|  3:2,3|2:1|    5.0|          3|
    |  2|          2|    2.0|          0|
    |  3|    2:12,16|    2.0|          2|
    |  4|    3:2,4,6|    3.0|          3|
    |  5|          2|    2.0|          0|
    |  6|  2:3,7|2:3|    4.0|          3|
    +---+-----------+-------+-----------+