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

Question

我有一个Dataset<Row>要转置的 Apache Spark。从这里的一些主题中，我很清楚这可以通过分组-透视-聚合来完成。但是我没有按照我需要的方式得到它。我有以下输入表：+-------+------+------+------+------+| 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]|

+------+-------------+

基本上，我构建了一个数组列，其中包含由列名及其值组成的结构。然后，我使用 explode 函数来展平这个数组。从那里，我们可以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|

+-----+------+---+-----+