在 pyspark 中旋转多个列以进行 PCA 准备

Question

我有大量数据要为 PCA 做准备。 它看起来如下所示，其中RESULT 列是我要分配给其他两列中的每一列的数值

RESULT  WINDOW
55.47   PM2_ArmA
195.3   PM3_ArmA
55.47   PM2_ArmA
66.94   PM3_ArmA
187.48  PM2_ArmA
298.99  PM2_ArmA
121.21  PM3_ArmA
153.44  PM2_ArmA
219.32  PM3_ArmA
289.01  PM2_ArmA
190.6   PM2_ArmA
287.57  PM3_ArmA
282.65  PM2_ArmA

预期输出：

PM2_ArmA    PM3_ArmA
55.47         195.3
55.47         66.94
187.48        121.21
298.99        219.32
153.44        287.57
289.01  
190.6   
282.65  

基本上，分配给WINDOW 列中每个名称的RESULT 值应该分配给它的名称，其中名称成为它自己的列。 乍一看，我以为这是一个简单的枢轴过程，但在 Pyspark 中执行此操作似乎更复杂。 有没有办法使用 pyspark 做到这一点？

我正在考虑的一种方法涉及以下步骤： 如果原始数据框是df，列RESULT 和WINDOW

1. 识别WINDOW 列中的唯一值并将它们放置在单独的一维数据框unq 中。将unq 转换为列表。
2. 创建多个数据帧，其中df1=df.filter(col('WINDOW')==unq[1]) 到dfn，其中dfn=df.filter(col('WINDOW')==unq[n])。为此将使用 for 循环
3. 我现在将拥有多个数据框，其中包含 RESULT 列和 WINDOW 其中WINDOW 列中的值将等于 unq[i].
4. 删除WINDOW 列并重命名RESULT 列 与unq[i] 横向联合所有数据帧。 （我不知道怎么 完成这部分）

我确信在 Pyspark 中有一种更优雅的方法。

Answer 1

您已经在编辑中描述了解决方案：

df=...input data...

#step 1
unq=list(map(lambda r:r['WINDOW'], df.select('WINDOW').distinct().collect()))

#step 2 and 3
dfs=[df.filter(f'WINDOW="{w}"').select('RESULT').rdd.map(lambda r: r[0]) \
  .zipWithIndex().map( lambda r: (r[1],r[0])).toDF(schema=['id', w])
  for w in unq]

#step 4
from functools import reduce
result=reduce(lambda df1, df2: df1.join(df2, ['id'], how='outer'), dfs).drop('id')

result.show()

输出：

+--------+--------+                                                             
|PM2_ArmA|PM3_ArmA|
+--------+--------+
|   55.47|   195.3|
|  282.65|    null|
|   190.6|    null|
|  289.01|    null|
|   55.47|   66.94|
|  298.99|  219.32|
|  187.48|  121.21|
|  153.44|  287.57|
+--------+--------+

通过使用RDD.zipWithIndex 为每个单个数据帧添加索引，然后使用索引列连接所有数据帧，可以将单个数据帧组合成一个数据帧。不幸的是，使用zipWithIndex 后跟外连接并不是最快的操作，并且会触发多次随机播放。