将结构类型列分解为pyspark中的两列键和值

Question

我有一个 pyspark df，其架构看起来像这样

 root
 |-- company: struct (nullable = true)
 |    |-- 0: string (nullable = true)
 |    |-- 1: string (nullable = true)
 |    |-- 10: string (nullable = true)
 |    |-- 100: string (nullable = true)
 |    |-- 101: string (nullable = true)
 |    |-- 102: string (nullable = true)
 |    |-- 103: string (nullable = true)
 |    |-- 104: string (nullable = true)
 |    |-- 105: string (nullable = true)
 |    |-- 106: string (nullable = true)
 |    |-- 107: string (nullable = true)
 |    |-- 108: string (nullable = true)
 |    |-- 109: string (nullable = true)

我希望这个数据框的最终格式看起来像这样

id    name
0     "foo"
1     "laa"
10    "bar"
100   "gee"
101   "yoo"
102    "koo"

而不是

  0        1         10       100      101        102 
"foo"    "laa"      "bar"    "gee"    "yoo"      "koo"

这是我使用 'col.*' 扩展得到的结果

我在这个链接中找到了答案 How to explode StructType to rows from json dataframe in Spark rather than to columns

但那是 scala spark 而不是 pyspark。我不熟悉 map reduce 概念，我自己将这里的脚本更改为 pyspark。

我在下面附加了一个类似架构和结构的示例数据框..

from pyspark.sql import *
Employee = Row('employee1', 'employee2', 'employee3', 'employee4', 'employee5')
Salaries = Row('100000', '120000', '140000', '160000', '160000')

departmentWithEmployees1 = Row(employees=[Employee, Salaries])

departmentsWithEmployees_Seq = [departmentWithEmployees1]
dframe = spark.createDataFrame(departmentsWithEmployees_Seq)
dframe.show()

这个dataframe的结构是这样的

root
 |-- employees: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- _1: string (nullable = true)
 |    |    |-- _2: string (nullable = true)
 |    |    |-- _3: string (nullable = true)
 |    |    |-- _4: string (nullable = true)
 |    |    |-- _5: string (nullable = true)

我想要的最终数据框是这样的

Firstname         Salary
employee1         10000
employee2         120000

Answer 1

这可以通过两个简单的选择语句来完成。

请注意，您提供的两个示例略有不同，在第二个示例中，结构 col 位于数组 col 中。

我将介绍更复杂的，但对于第一个（和原始 df），您可以跳过第一个 select 语句。

dframe\
.selectExpr('employees[0] AS `key`', 'employees[1] AS `value`')\
.select(
    F.explode(F.map_from_arrays(F.array('key.*'),F.array('value.*'))
             ).alias('Firstname','Salary')
)

我将尝试解释下面的逻辑。

root
 |-- employees: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- _1: string (nullable = true)
 |    |    |-- _2: string (nullable = true)
 |    |    |-- _3: string (nullable = true)
 |    |    |-- _4: string (nullable = true)
 |    |    |-- _5: string (nullable = true)

查看上面的架构你需要做的是：

1) 展平第一个数组col以暴露结构

2) 将两个struct cols 变成两个array cols，创建一个map cols 和map_from_arrays() cols 并爆炸。

map_from_arrays() 从两个 array 列的同一位置获取一个元素（想想 Python zip()）。

如果这有帮助，请告诉我！

Answer 2

首先使用element_at 获取firstname 和salary 列，然后使用F.array 和F.arrays_zip 列在explode 之前将它们从结构转换为数组，然后select 全部爆炸压缩列。

from pyspark.sql import functions as F
dframe.withColumn("firstname", F.element_at("employees", 1))\
.withColumn("salary",F.element_at("employees",2))\
.drop("employees")\
.withColumn("firstname",F.array("firstname.*"))\
.withColumn("salary", F.array("salary.*"))\
.withColumn("zip",F.explode(F.arrays_zip("firstname","salary")))\
.select("zip.*").show(truncate=False)

+---------+------+
|firstname|salary|
+---------+------+
|employee1|100000|
|employee2|120000|
|employee3|140000|
|employee4|160000|
|employee5|160000|
+---------+------+