スパーク - 共通の列ID

Question

の列を作成するために、複数の行を追加は、私は現在、この私がフォームでそれを変換するために探しています

|col_id|r_id_1|r_id_2|r_id_3| 
| 1 | a1 | b1 | c1 | 
| 1 | a2 | b2 | c2 | 
| 2 | a3 | b3 | c3 | 
| 2 | a4 | b4 | c4 | 

|col_id|r_id_1|r_id_2|r_id_3|r_id_1|r_id_2|r_id_3| 
| 1 | a1 | b1 | c1 | a2 | b2 | c2 | 
| 2 | a3 | b3 | c3 | a4 | b4 | c4 | 

このようにのように見えるのデータフレームを持って、列IDが1の2行があり、col_idでグループ化されており、既存の行を使用して新しい列を生成するはずです。 注。各列idの行数は同じです。

Answer 1

これは、実行する必要があります。

import org.apache.spark.sql.expressions._ 
import org.apache.spark.sql.functions._ 

val df = Seq(
    (1, "a1", "b1", "c1"), 
    (1, "a2", "b2", "c2"), 
    (2, "a3", "b3", "c3"), 
    (2, "a4", "b4", "c4") 
).toDF("col_id", "r_id_1", "r_id2", "r_id_3") 

val cols = df.columns.tail 

df 
    .withColumn("rn", 
    row_number().over(Window.partitionBy("col_id").orderBy("r_id_1"))) 
.flatMap { row => row.getValuesMap[String](cols).map { 
    case (c, t) => (row.getAs[Int]("col_id"), s"${c}_${row.getAs[Int]("rn")}", t) }} 
.groupBy("_1") 
.pivot("_2") 
.agg(first("_3")) 
.show 

+---+-------+-------+--------+--------+--------+--------+      
| _1|r_id2_1|r_id2_2|r_id_1_1|r_id_1_2|r_id_3_1|r_id_3_2| 
+---+-------+-------+--------+--------+--------+--------+ 
| 1|  b1|  b2|  a1|  a2|  c1|  c2| 
| 2|  b3|  b4|  a3|  a4|  c3|  c4| 
+---+-------+-------+--------+--------+--------+--------+