如何在任意列上透视？

Question

我使用Apache Spark 2.2.0和Scala。

我以这个问题为指导，在不使用pivot函数的情况下透视数据帧。

我需要在不使用pivot函数的情况下透视数据帧，因为我有非数字数据，而< code>pivot只对数字数据使用聚合函数，如< code>sum 、< code>min 、< code>max。我想在< code>pivot聚合中使用一个非数字列。

这是我的数据:

+---+-------------+----------+-------------+----------+-------+
|Qid|     Question|AnswerText|ParticipantID|Assessment| GeoTag|
+---+-------------+----------+-------------+----------+-------+
|  1|Question1Text|       Yes|       abcde1|         0|(x1,y1)|
|  2|Question2Text|        No|       abcde1|         0|(x1,y1)|
|  3|Question3Text|         3|       abcde1|         0|(x1,y1)|
|  1|Question1Text|        No|       abcde2|         0|(x2,y2)|
|  2|Question2Text|       Yes|       abcde2|         0|(x2,y2)|
+---+-------------+----------+-------------+----------+-------+

我希望它按< code>ParticipantID、< code>Assessment和< code>GeoTag标记分组，并在< code>Question列上“pivot”并从< code>AnswerText列中取值。最后，输出应该如下所示:

+-------------+-----------+----------+-------+-----+----- +
|ParticipantID|Assessment |GeoTag    |Qid_1  |Qid_2|Qid_3 |
+-------------+-----------+----------+-------+-----+------+
|abcde1       |0          |(x1,y1)   |Yes    |No   |3     |
|abcde2       |0          |(x2,y2)   |No     |Yes  |null  |
+-------------+-----------+----------+-------+-----+------+

我试过这个：

val questions: Array[String] = df.select("Q_id")
      .distinct()
      .collect()
      .map(_.getAs[String]("Q_id"))
      .sortWith(_<_)

val df2: DataFrame = questions.foldLeft(df) {
      case (data, question) => data.selectExpr("*", s"IF(Q_id = '$question', AnswerText, 0) AS $question")
    }

[后跟分组依据表达式]

但是我得到了以下错误，这一定与最终语句AS$问题的语法有关

17/12/08 16:13:12 INFO SparkSqlParser: Parsing command: *
17/12/08 16:13:12 INFO SparkSqlParser: Parsing command: IF(Q_id_string_new_2 = '101_Who_is_with_you_right_now?', AnswerText, 0) AS 101_Who_is_with_you_right_now?


extraneous input '?' expecting <EOF>(line 1, pos 104)

== SQL ==
IF(Q_id_string_new_2 = '101_Who_is_with_you_right_now?', AnswerText, 0) AS 101_Who_is_with_you_right_now?
--------------------------------------------------------------------------------------------------------^^^

org.apache.spark.sql.catalyst.parser.ParseException: 
extraneous input '?' expecting <EOF>(line 1, pos 104)

== SQL ==
IF(Q_id_string_new_2 = '101_Who_is_with_you_right_now?', AnswerText, 0) AS 101_Who_is_with_you_right_now?
--------------------------------------------------------------------------------------------------------^^^

    at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala:217)

你知道我哪里出了问题吗？有更好的方法吗？如果有必要，我考虑在Spark之外恢复使用Panda和Python，但如果可能的话，我宁愿在相同的框架内编写所有代码。

Answer 1

只需注意@user8371915接受的答案，以使查询更快。

有一种方法可以避免使用标头生成问题的昂贵扫描。

只需生成标题（在同一作业和阶段中！）然后是列上的枢轴。

// It's a simple and cheap map-like transformation
val qid_header = input.withColumn("header", concat(lit("Qid_"), $"Qid"))
scala> qid_header.show
+---+-------------+----------+-------------+----------+-------+------+
|Qid|     Question|AnswerText|ParticipantID|Assessment| GeoTag|header|
+---+-------------+----------+-------------+----------+-------+------+
|  1|Question1Text|       Yes|       abcde1|         0|(x1,y1)| Qid_1|
|  2|Question2Text|        No|       abcde1|         0|(x1,y1)| Qid_2|
|  3|Question3Text|         3|       abcde1|         0|(x1,y1)| Qid_3|
|  1|Question1Text|        No|       abcde2|         0|(x2,y2)| Qid_1|
|  2|Question2Text|       Yes|       abcde2|         0|(x2,y2)| Qid_2|
+---+-------------+----------+-------------+----------+-------+------+

将标头作为数据集的一部分，让我们进行透视。

val solution = qid_header
  .groupBy('ParticipantID, 'Assessment, 'GeoTag)
  .pivot('header)
  .agg(first('AnswerText))
scala> solution.show
+-------------+----------+-------+-----+-----+-----+
|ParticipantID|Assessment| GeoTag|Qid_1|Qid_2|Qid_3|
+-------------+----------+-------+-----+-----+-----+
|       abcde1|         0|(x1,y1)|  Yes|   No|    3|
|       abcde2|         0|(x2,y2)|   No|  Yes| null|
+-------------+----------+-------+-----+-----+-----+

Answer 2

当< code>$question将问题变量的值代入SQL语句时，您会得到一个带“？”的列名在它的SQL中。<代码>？在列名中不是有效字符，因此您必须至少使用反斜杠来引用:

s"IF(Q_id = '$question', AnswerText, 0) AS `$question`"

或者使用< code > select /< code > with column :

import org.apache.spark.sql.functions.when

case (data, question) => 
  data.withColumn(question, when($"Q_id" === question, $"AnswerText"))

或者首先使用< code>regexp_replace对字符串进行santize。

我需要在不使用pivot函数的情况下透视数据帧，因为我有非数字数据和df.pivot只对数字数据使用sum、min、max等聚合函数。

您可以先使用＜code＞来：如何使用pivot并计算非数值列的平均值（面向AnalysisException“不是数值列”）？

data.groupBy($"ParticipantID", $"Assessment", $"GeoTag")
  .pivot($"Question", questions).agg(first($"AnswerText"))