pandas - 如何对 pyspark dataframe 中的单列进行 reshape 操作？

Question

我有一个很长的 pyspark 数据框，如下所示:

+------+
|number|
+------+
|12.4  |
|13.4  |
|42.3  |
|33.4  |
|42.3  |
|32.4  |
|44.2  |
|12.3  |
|45.4  |
+------+

理想情况下，我希望将其 reshape 为 nxn 矩阵，其中 n 是 sqrt(pyspark 数据帧的长度)。

虽然有一个解决方案，将其转换为 numpy 数组，然后将其 reshape 为 nxn 矩阵，但我希望在 pyspark 中完成。因为我的数据超长(大约1亿行)。

所以我正在寻找的预期输出是这样的:

+------+------+------+
|12.4  | 13.4 | 42.3 |
|33.4  | 42.3 | 32.4 |
|44.2  | 12.3 | 45.4 |
+------+------+------+

虽然我能够通过将其转换为 pandas 然后转换为 numpy 然后执行 reshape 操作来正确完成此操作。但我想在 Pyspark 本身中进行这种转换。因为下面的代码只适用于几千行。

covarianceMatrix_pd = covarianceMatrix_df.toPandas()
nrows = np.sqrt(len(covarianceMatrix_pd))
covarianceMatrix_pd = covarianceMatrix_pd.to_numpy().reshape((int(nrows),int(nrows)))
covarianceMatrix_pd

Answer 1

实现此目的的一种方法是在获得数据帧的计数后将 row_number 与数据透视一起使用:

from pyspark.sql import functions as F, Window
from math import sqrt

---

c = int(sqrt(df.count())) #this gives 3
rnum = F.row_number().over(Window.orderBy(F.lit(1)))

out = (df.withColumn("Rnum",((rnum-1)/c).cast("Integer"))
 .withColumn("idx",F.row_number().over(Window.partitionBy("Rnum").orderBy("Rnum")))
.groupby("Rnum").pivot("idx").agg(F.first("number")))

---

out.show()

+----+----+----+----+
|Rnum|   1|   2|   3|
+----+----+----+----+
|   0|12.4|13.4|42.3|
|   1|33.4|42.3|32.4|
|   2|44.2|12.3|45.4|
+----+----+----+----+