Pyspark – 将多个数组列拆分成行

在大数据处理中，经常需要处理数组列。然而，有时候需要将多个数组列拆分成行，并根据其索引位置组合为新行。本文将介绍如何在Pyspark中实现此操作。

准备工作

在开始之前，我们需要确保已经安装好Pyspark和相关依赖，并准备好输入数据。

我们将使用以下示例数据：

+---+--------+-----------+--------+
| id|    arr1|       arr2|    arr3|
+---+--------+-----------+--------+
|  1|[1, 2, 3]|[a, b, c, d]|[x, y, z]|
|  2|[4, 5, 6]|[e, f, g, h]|[p, q, r]|
+---+--------+-----------+--------+

实现拆分操作

首先，我们需要使用posexplode_arrays函数将数组列转换为行。该函数会将输入数组中每个元素拆分成一行，并分配一个新的行索引。例如，我们将使用以下代码生成一个新的DataFrame：

from pyspark.sql.functions import posexplode, posexplode_outer, expr

df = spark.createDataFrame([(1, [1, 2, 3], ['a', 'b', 'c', 'd'], ['x', 'y', 'z']),
                            (2, [4, 5, 6], ['e', 'f', 'g', 'h'], ['p', 'q', 'r'])],
                           ['id', 'arr1', 'arr2', 'arr3'])

df_posexploded = df.selectExpr("id", "posexplode(arr1)", "posexplode(arr2)", "posexplode(arr3)")

df_posexploded.show()

输出结果如下：

+---+----+---+----+---+----+----+
| id|pos1|col1|pos2|col2|pos3|col3|
+---+----+---+----+---+----+----+
|  1|   0|  1|   0|  a|   0|   x|
|  1|   1|  2|   1|  b|   1|   y|
|  1|   2|  3|   2|  c|   2|   z|
|  2|   0|  4|   0|  e|   0|   p|
|  2|   1|  5|   1|  f|   1|   q|
|  2|   2|  6|   2|  g|   2|   r|
+---+----+---+----+---+----+----+

接下来，我们需要使用pivot函数将每个索引位置的元素组合为新行。例如，我们将使用以下代码将三个拆分的列组合为新行：

from pyspark.sql.functions import concat_ws

df_pivoted = df_posexploded.groupBy("id", "pos1").agg(concat_ws(", ", "col1", "col2", "col3").alias("arr"))

df_pivoted.show()

输出结果如下：

+---+----+-------+
| id|pos1|    arr|
+---+----+-------+
|  1|   0| 1, a, x|
|  1|   1| 2, b, y|
|  1|   2| 3, c, z|
|  2|   0| 4, e, p|
|  2|   1| 5, f, q|
|  2|   2| 6, g, r|
+---+----+-------+

现在，我们已经将拆分的数组列转换为单个DataFrame，并将其组合为新行。

总结

在本文中，我们介绍了如何在Pyspark中将多个数组列拆分成行，并将其组合为新行。我们首先使用posexplode_arrays函数将数组列转换为行，然后使用pivot函数将每个索引位置的元素组合为新行。这是处理数组列的有用技巧，可以帮助简化大数据处理中的操作。