PySpark爆炸JSON字符串成多列

Question

我有一个字符串数据类型列的数据框架。该字符串表示返回json的api请求。

df = spark.createDataFrame([
("[{original={ranking=1.0, input=top3}, response=[{to=Sam, position=guard}, {to=John, position=center}, {to=Andrew, position=forward}]}]",1)], 
"col1:string, col2:int")
df.show()

生成如下数据帧:

+--------------------+----+
|                col1|col2|
+--------------------+----+
|[{original={ranki...|   1|
+--------------------+----+

输出我希望有col2和来自响应的两个附加列。Col3将捕获玩家的名字，用to=表示，col4的位置用position=表示。数据框现在有三行，因为有三个玩家。例子:

+----+------+-------+
|col2|  col3|   col4|
+----+------+-------+
|   1|   Sam|  guard|
|   1|  John| center|
|   1|Andrew|forward|
+----+------+-------+

我读到我可以利用这样的东西:

df.withColumn("col3",explode(from_json("col1")))

然而，我不确定如何爆炸给我想要两列而不是一个，需要模式。

注意，我可以使用json_dumps修改响应，使其只返回字符串的响应部分或…

[{to=Sam, position=guard}, {to=John, position=center}, {to=Andrew, position=forward}]}]

Answer 1

如果您像前面提到的那样简化输出，您可以定义一个简单的JSON模式并将JSON字符串转换为StructType并读取每个字段

输入>

df = spark.createDataFrame([("[{'to': 'Sam', 'position': 'guard'},{'to': 'John', 'position': 'center'},{'to': 'Andrew', 'position': 'forward'}]",1)], "col1:string, col2:int")
# +-----------------------------------------------------------------------------------------------------------------+----+
# |col1                                                                                                             |col2|
# +-----------------------------------------------------------------------------------------------------------------+----+
# |[{'to': 'Sam', 'position': 'guard'},{'to': 'John', 'position': 'center'},{'to': 'Andrew', 'position': 'forward'}]|1   |
# +-----------------------------------------------------------------------------------------------------------------+----+

这是变换

from pyspark.sql import functions as F
from pyspark.sql import types as T
schema = T.ArrayType(T.StructType([
T.StructField('to', T.StringType()),
T.StructField('position', T.StringType())
]))
(df
.withColumn('temp', F.explode(F.from_json('col1', schema=schema)))
.select(
F.col('col2'),
F.col('temp.to').alias('col3'),
F.col('temp.position').alias('col4'),
)
.show()
)
# Output
# +----+------+-------+
# |col2|  col3|   col4|
# +----+------+-------+
# |   1|   Sam|  guard|
# |   1|  John| center|
# |   1|Andrew|forward|
# +----+------+-------+