如何按性别分组并按每个组的职位加入？

Question

我尝试了多种方法来解决以下问题：

 Gender, Age, Value
 1,      20,  21
 2,      23   22
 1,      26,  23
 2,      29,  24

进入

   Male_Age, Male_Value, Female_Age,  Female_Value
     20          21         23           22
     26          23         29           24

我需要做的是按性别分组，而不是使用像（sum、count、avg）这样的聚合，我需要创建 List[age] 和 List[value]。这应该是可能的，因为我使用的是允许功能操作的数据集。

如果男性和女性的行数不一样，列应该填nulls。

我尝试的一种方法是使用其他数据框的列创建一个新的新数据框，如下所示：

df .select(male.select("sex").where('sex === 1).col("sex"),
female.select("sex").where('sex === 2).col("sex"))

但是，这会奇怪地产生如下输出：

sex,    sex,
 1,       1
 2,       2
 1,       1
 2,       2

我看不出这怎么可能。

我也尝试过使用数据透视，但它迫使我在分组后聚合：

df.withColumn("sex2", df.col("sex"))
.groupBy("性")
.pivot("sex2") .agg( sum('value').as("mean"),
stddev('value).as("std.dev") ) .show()

|sex|    1.0_mean|   1.0_std. dev|   2.0_mean|    2.0_std. dev|
|1.0|0.4926065526|   1.8110632697|           |                |
|2.0|            |               |0.951250372|1.75060275400785|

以下代码可以满足我在 Oracle SQL 中的需要，因此我认为在 Spark SQL 中也应该可以实现...

drop table mytable

CREATE TABLE mytable 
( gender number(10) NOT NULL,
  age number(10) NOT NULL,   
  value number(10) );

 insert into mytable values (1,20,21); 
 insert into mytable values(2,23,22); 
 insert into mytable values (1,26,23); 
 insert into mytable values (2,29,24); 
 insert into mytable values (1,30,25);

 select * from mytable;


SELECT A.VALUE AS MALE, 
       B.VALUE AS FEMALE 
FROM 
(select value, rownum RN from mytable where gender = 1) A 
FULL OUTER JOIN 
(select value, rownum RN from mytable where gender = 2) B
ON A.RN = B.RN

Answer 1

下面应该会给你结果。

val df = Seq(
  (1,      20,  21),
  (2,      23,   22),
  (1,      26,  23),
  (2,      29,  24)
).toDF("Gender", "Age", "Value")

scala> df.show
+------+---+-----+
|Gender|Age|Value|
+------+---+-----+
|     1| 20|   21|
|     2| 23|   22|
|     1| 26|   23|
|     2| 29|   24|
+------+---+-----+

// Gender 1 = Male
// Gender 2 = Female

import org.apache.spark.sql.expressions.Window
val byGender = Window.partitionBy("gender").orderBy("gender")

val males = df
  .filter("gender = 1")
  .select($"age" as "male_age",
          $"value" as "male_value",
          row_number() over byGender as "RN")

scala> males.show
+--------+----------+---+
|male_age|male_value| RN|
+--------+----------+---+
|      20|        21|  1|
|      26|        23|  2|
+--------+----------+---+

val females = df
  .filter("gender = 2")
  .select($"age" as "female_age",
          $"value" as "female_value",
          row_number() over byGender as "RN")

scala> females.show
+----------+------------+---+
|female_age|female_value| RN|
+----------+------------+---+
|        23|          22|  1|
|        29|          24|  2|
+----------+------------+---+

scala> males.join(females, Seq("RN"), "outer").show
+---+--------+----------+----------+------------+
| RN|male_age|male_value|female_age|female_value|
+---+--------+----------+----------+------------+
|  1|      20|        21|        23|          22|
|  2|      26|        23|        29|          24|
+---+--------+----------+----------+------------+

Answer 2

给定一个名为df 的DataFrame 包含gender、age 和value 列，您可以这样做：

df.groupBy($"gender")
  .agg(collect_list($"age"), collect_list($"value")).rdd.map { row =>
     val ages: Seq[Int] = row.getSeq(1)
     val values: Seq[Int] = row.getSeq(2)
     (row.getInt(0), ages.head, ages.last, values.head, values.last)
  }.toDF("gender", "male_age", "female_age", "male_value", "female_value")

这使用了非常有用的 Spark functions library 中的 collect_list 聚合函数来聚合您想要的值。 （如您所见，还有一个collect_set。）

在那之后，我不知道有任何更高级别的 DataFrame 函数可以将这些列状数组扩展为它们自己的单独列，所以我回退到我们祖先使用的低级别 RDD API。我只是将所有内容扩展为Tuple，然后将其转回DataFrame。上面的评论者提到了我没有解决的极端案例；使用 headOption 和 tailOption 之类的函数可能在那里有用。但这应该足以让你动起来。