如何在Pyspark中将列表拆分为多列？

Question

我有:

key   value
a    [1,2,3]
b    [2,3,4]

我想要：

key value1 value2 value3
a     1      2      3
b     2      3      4

似乎在scala中我可以写:< code>df.select($"value。_1 "，$ "值。_2 "，$ "值。_3")，但这在python中是不可能的。

那么有没有好的办法呢？

Answer 1

我需要将712维数组取消列出到列中，以便将其写入csv。我首先使用@MaFF的解决方案来解决我的问题，但这似乎会导致很多错误和额外的计算时间。我不确定是什么原因造成的，但我使用了一种不同的方法，大大减少了计算时间（22分钟与超过4小时相比）！

@MaFF的方法:

length = len(dataset.head()["list_col"])
dataset = dataset.select(dataset.columns + [dataset["list_col"][k] for k in range(length)])

我用的是:

dataset = dataset.rdd.map(lambda x: (*x, *x["list_col"])).toDF()

如果有人知道是什么导致了计算时间的差异，请告诉我！我怀疑在我的例子中，瓶颈在于调用＜code＞head（）

Answer 2

我想将大小列表（数组）的情况添加到pault答案中。

如果我们的列包含中型数组（或大型数组），仍然可以将它们拆分为列。

from pyspark.sql.types import *          # Needed to define DataFrame Schema.
from pyspark.sql.functions import expr   

# Define schema to create DataFrame with an array typed column.
mySchema = StructType([StructField("V1", StringType(), True),
                       StructField("V2", ArrayType(IntegerType(),True))])

df = spark.createDataFrame([['A', [1, 2, 3, 4, 5, 6, 7]], 
                            ['B', [8, 7, 6, 5, 4, 3, 2]]], schema= mySchema)

# Split list into columns using 'expr()' in a comprehension list.
arr_size = 7
df = df.select(['V1', 'V2']+[expr('V2[' + str(x) + ']') for x in range(0, arr_size)])

# It is posible to define new column names.
new_colnames = ['V1', 'V2'] + ['val_' + str(i) for i in range(0, arr_size)] 
df = df.toDF(*new_colnames)

结果是：

df.show(truncate= False)

+---+---------------------+-----+-----+-----+-----+-----+-----+-----+
|V1 |V2                   |val_0|val_1|val_2|val_3|val_4|val_5|val_6|
+---+---------------------+-----+-----+-----+-----+-----+-----+-----+
|A  |[1, 2, 3, 4, 5, 6, 7]|1    |2    |3    |4    |5    |6    |7    |
|B  |[8, 7, 6, 5, 4, 3, 2]|8    |7    |6    |5    |4    |3    |2    |
+---+---------------------+-----+-----+-----+-----+-----+-----+-----+

Answer 3

这取决于您的“列表”的类型:

>

如果它是ArrayType（）类型：

df = hc.createDataFrame(sc.parallelize([['a', [1,2,3]], ['b', [2,3,4]]]), ["key", "value"])
df.printSchema()
df.show()
root
 |-- key: string (nullable = true)
 |-- value: array (nullable = true)
 |    |-- element: long (containsNull = true)

你可以像使用 [] 的 python 一样访问这些值：

df.select("key", df.value[0], df.value[1], df.value[2]).show()
+---+--------+--------+--------+
|key|value[0]|value[1]|value[2]|
+---+--------+--------+--------+
|  a|       1|       2|       3|
|  b|       2|       3|       4|
+---+--------+--------+--------+

+---+-------+
|key|  value|
+---+-------+
|  a|[1,2,3]|
|  b|[2,3,4]|
+---+-------+

如果它是结构类型（）类型：（也许您通过读取JSON来构建数据帧）

df2 = df.select("key", psf.struct(
        df.value[0].alias("value1"), 
        df.value[1].alias("value2"), 
        df.value[2].alias("value3")
    ).alias("value"))
df2.printSchema()
df2.show()
root
 |-- key: string (nullable = true)
 |-- value: struct (nullable = false)
 |    |-- value1: long (nullable = true)
 |    |-- value2: long (nullable = true)
 |    |-- value3: long (nullable = true)

+---+-------+
|key|  value|
+---+-------+
|  a|[1,2,3]|
|  b|[2,3,4]|
+---+-------+

您可以使用< code>*直接“拆分”列:

df2.select('key', 'value.*').show()
+---+------+------+------+
|key|value1|value2|value3|
+---+------+------+------+
|  a|     1|     2|     3|
|  b|     2|     3|     4|
+---+------+------+------+