pyspark分解嵌套列表

yks3o0rb 于 2021-05-24 发布在 Spark

关注(0)|答案(1)|浏览(609)

我有以下数据框，我想分解values列，以便每个值都在单独的列中：

id | values
-----------------------
1  | '[[532,969020406,89],[216,969100125,23],[169,39356140000,72],[399,14407358500,188],[377,13761937166.6667,24]]'
2 | '[[532,969020406,89]]'

请注意，“值”列下的列表可以有不同的长度，并且它们是字符串数据类型。
所需的表应如下所示：

id | v11 | v12 | v13 | v21 | v22... 
--------------------------------------
1  | 532 | 969020406 | 89 | 216 | 969100125...
2 | 532 | 969020406 | 89 | Null | Null...

我试图指定模式，并使用from\ json方法创建数组，然后分解它，但遇到了问题，即任何模式似乎都不适合我的数据

json_schema =  types.StructType([types.StructField('array', types.StructType([ \
    types.StructField("v1",types.StringType(),True), \
    types.StructField("v2",types.StringType(),True), \
    types.StructField("v3",types.StringType(),True)
  ]))])

json_schema = types.ArrayType(types.StructType([ \
    types.StructField("v1",types.StringType(),True), \
    types.StructField("v2",types.StringType(),True), \
    types.StructField("v3",types.StringType(),True)
  ]))

json_schema = types.ArrayType(types.ArrayType(types.IntegerType()))

df.select('id', F.from_json('values', schema=json_schema)).show()

程序仅返回空值或空数组：[，，，]
我还遇到以下错误：structtype不能接受type<class'str'>
pyspark推断的输入数据的架构：

root
 |-- id: integer (nullable = true)
 |-- values: string (nullable = true)

任何帮助都将不胜感激。

apache-spark pyspark

来源：https://stackoverflow.com/questions/64139158/pyspark-explode-nested-list

1条答案

按热度按时间

b1payxdu1#

对于spark 2.4+，可以使用拆分和变换的组合将字符串转换为二维数组。然后可以将此数组的单个条目分别转换为列。

from pyspark.sql import functions as F

df2 = df.withColumn("parsed_values", F.expr("transform(split(values, '\\\\],\\\\['), " +
           "c ->  transform(split(c, ','), d->regexp_replace(d,'[\\\\[\\\\]]','')))"))\
    .withColumn("length", F.size("parsed_values"))

max_length = df2.agg(F.max("length")).head()["max(length)"]
``` `df2` 现在有了结构

以及 `max_length` 包含一行中的最大条目数（示例数据为5）。 `parsed_value[0][1]` 将返回第一个条目的第二个子条目。这将是 `969020406` 对于示例数据。
第二步是将嵌套数组转换为列。

cols = [F.col('parsed_values').getItem(x).getItem(y).alias("v{}{}".format(x+1,y+1))
for x in range(0, max_length) for y in range(0,3)]

df2.select([F.col('id')] + cols).show()

输出：

+---+---+---------+---+----+---------+----+----+-----------+----+----+-----------+----+----+----------------+----+
| id|v11| v12|v13| v21| v22| v23| v31| v32| v33| v41| v42| v43| v51| v52| v53|
+---+---+---------+---+----+---------+----+----+-----------+----+----+-----------+----+----+----------------+----+
| 1|532|969020406| 89| 216|969100125| 23| 169|39356140000| 72| 399|14407358500| 188| 377|13761937166.6667| 24|
| 2|532|969020406| 89|null| null|null|null| null|null|null| null|null|null| null|null|
+---+---+---------+---+----+---------+----+----+-----------+----+----+-----------+----+----+----------------+----+

如果有办法确定 `max_length` 无需在完整数据上找到最大值，例如，如果该值事先已知。

赞(0）回复(0）举报 2021-05-25

我来回答

pyspark分解嵌套列表

1条答案

相关问题

热门标签

最新问答