java - 如何在 Java 中转置 Apache Spark 数据集

Question

我有一个 Apache Spark Dataset<Row>我想转置它。从这里的一些主题中，我清楚地知道这可以通过分组-旋转-聚合来完成。但我没有以我需要的方式得到它。我有以下输入表:

+-------+------+------+------+------+
| rho_0 | rho_1| rho_2|rho_3 | names|
+-------+------+------+------+------+
|  1    | 0.89 | 0.66 | 0.074|  rho |
|  1    | 0.89 | 0.66 | 0.074|absRho|
|  0    |  1   | 2    |  3   | lag  |
+-------+------+------+------+------+

我需要的是

+-------+------+------+
| rho   |absRho| lag  |
+-------+------+------+
|  1    | 1    |  0   |
|  0.89 | 0.89 |  1   |
|  0.66 | 0.66 |  2   |
|  0.074| 0.074|  3   |
+-------+------+------+

我尝试过类似的方法

Dataset<Row> transposed = coll.groupBy().pivot("names").min("rho_0");

但这不起作用。调用groupBy使用输入中的一系列列也不起作用。我找到了一个我不喜欢的解决方法:

Dataset<Row> transposed = coll.groupBy().pivot("names").min("rho_0")
for (int i = 1; i < nlags; i++) {
    transposed = transposed.union(coll.groupBy().pivot("names").min("rho_" + i));
}

但是它真的很慢并且不应该以这种方式实现。你有什么建议吗？提前致谢!

Answer 1

不幸的是，spark 中没有内置函数可以做到这一点。有一个使用pivot的解决方案，但是您需要先“分解”数据帧。它应该比基于联合的解决方案快得多。

在 scala 中，它会如下所示。我在下面添加了一个java版本。

// scala
val cols = df.columns
  .filter(_ != "names")
  .map(n => struct(lit(n) as "c", col(n) as "v"))
val exploded_df = df.select(col("names"), explode(array(cols : _*)))

// java
Column[] cols = Arrays
    .stream(df.columns())
    .filter(x -> ! x.equals("names"))
    .map(n -> struct(lit(n).alias("c"), col(n).alias("v")))
    .toArray(Column[]::new);
Dataset<Row> exploded_df = df.select(col("names"), explode(array(cols)));

exploded_df.show();
+------+-------------+
| names|          col|
+------+-------------+
|   rho|    [rho_0,1]|
|   rho| [rho_1,0.89]|
|   rho| [rho_2,0.66]|
|   rho|[rho_3,0.074]|
|absRho|    [rho_0,1]|
|absRho| [rho_1,0.89]|
|absRho| [rho_2,0.66]|
|absRho|[rho_3,0.074]|
|   lag|    [rho_0,0]|
|   lag|    [rho_1,1]|
|   lag|    [rho_2,2]|
|   lag|    [rho_3,3]|
+------+-------------+

基本上，我构建了一个数组列，其中包含由列名称及其值组成的结构。然后，我使用爆炸函数来展平这个数组。从那里，我们可以像往常一样使用pivot;-)

// scala and java
exploded_df
  .groupBy(col("col.c"))
  .pivot("names")
  .agg(first(col("col.v")))
  .orderBy("c")
  .show();
+-----+------+---+-----+
|    c|absRho|lag|  rho|
+-----+------+---+-----+
|rho_0|     1|  0|    1|
|rho_1|  0.89|  1| 0.89|
|rho_2|  0.66|  2| 0.66|
|rho_3| 0.074|  3|0.074|
+-----+------+---+-----+