使用Apache Spark中的pyspark进行Dataframe转置

Question

我有一个数据框 df，结构如下：

+-----+-----+-----+-------+
|  s  |col_1|col_2|col_...|
+-----+-----+-----+-------+
| f1  |  0.0|  0.6|  ...  |
| f2  |  0.6|  0.7|  ...  |
| f3  |  0.5|  0.9|  ...  |
|  ...|  ...|  ...|  ...  |

我想要计算这个数据框的转置，使其看起来像

+-------+-----+-----+-------+------+
|  s    | f1  | f2  | f3    |   ...|
+-------+-----+-----+-------+------+
|col_1  |  0.0|  0.6|  0.5  |   ...|
|col_2  |  0.6|  0.7|  0.9  |   ...|
|col_...|  ...|  ...|  ...  |   ...|

我尝试了这两个解决方案，但都返回数据框没有指定的方法：

方法1：

 for x in df.columns:
    df = df.pivot(x)

方法二：

df = sc.parallelize([ (k,) + tuple(v[0:]) for k,v in df.items()]).toDF()

我该怎么修复这个问题。

- Mehdi Ben Hamida

---

你可以在这里查看 https://dev59.com/QJXfa4cB1Zd3GeqPemt9 - Avishek Bhattacharya

Answer 1

如果数据足够小可以转置（不需要聚合操作）, 你可以将其转换为Pandas的 DataFrame：

df = sc.parallelize([
    ("f1", 0.0, 0.6, 0.5),
    ("f2", 0.6, 0.7, 0.9)]).toDF(["s", "col_1", "col_2", "col_3"])

df.toPandas().set_index("s").transpose()
s       f1   f2
col_1  0.0  0.6
col_2  0.6  0.7
col_3  0.5  0.9

如果数据太大，Spark 就无法帮助了。Spark DataFrame 按行分发数据(尽管在本地使用列式存储)，因此单个行的大小受本地内存限制。

- Alper t. Turker

---

在将数据转换回 Spark Dataframe 之前，您可能希望先重置索引，以避免在行中丢失列名。您可以使用命令"reset_index"来完成此操作。例如：df.toPandas().set_index("s").transpose().reset_index() - lfvv

我们如何将结果再次转换为数据帧？ - Aspirant

spark.createDataFrame(result)。 - Alper t. Turker

Answer 2

您可以尝试使用Databricks的Koalas。Koalas类似于Pandas，但是专为分布式处理而设计，并且在Pyspark中可用（至少从3.0.0版本开始）。

kdf = df.to_koalas()
kdf_t = kdf.transpose()
df_T = kdf_t.to_spark()

编辑：为了高效地访问Koalas，您需要定义分区，否则可能会导致严重的性能下降。

- s510