apache spark中的聚合函数答案

【问题标题】：aggregate function in apache sparkapache spark中的聚合函数
【发布时间】：2018-01-06 15:39:09
【问题描述】：

我需要基于 1 分钟的时间间隔聚合一个数据集。当我尝试这个时，它会抛出错误：

我的数据集是这样的

scala> newVX.show
+--------------------+-----+
|            datetime|value|
+--------------------+-----+
|2017-07-31 10:53:...| 0.26|
|2017-07-31 10:53:...| 0.81|
|2017-07-31 09:45:...| 0.42|
|2017-07-31 09:44:...|0.008|
|2017-07-31 09:37:...| 0.14|
|2017-07-31 09:35:...|0.365|
|2017-07-31 09:34:...|0.485|
|2017-07-31 09:33:...| 0.49|
|2017-07-31 09:28:...| 1.15|
|2017-07-31 09:27:...|0.325|
|2017-07-31 09:24:...|0.845|
|2017-07-31 09:24:...|0.045|
|2017-07-31 09:23:...|0.015|
|2017-07-31 09:20:...| 0.45|
|2017-07-31 09:20:...| 0.05|
|2017-07-31 09:19:...| 0.14|
|2017-07-31 09:18:...| 0.24|
|2017-07-31 09:12:...|0.125|
|2017-07-31 09:11:...|  0.3|
|2017-07-31 09:11:...| 0.13|
+--------------------+-----+


scala> newVX.groupBy("datetime","60 seconds").agg(avg("value")).show

org.apache.spark.sql.AnalysisException：无法解析列名“60 秒” 其中（日期时间，值）；在 org.apache.spark.sql.Dataset$$anonfun$resolve$1.apply(Dataset.scala:216) 在 org.apache.spark.sql.Dataset$$anonfun$resolve$1.apply(Dataset.scala:216) 在 scala.Option.getOrElse(Option.scala:121) 在 org.apache.spark.sql.Dataset.resolve(Dataset.scala:215) 在 org.apache.spark.sql.Dataset$$anonfun$groupBy$2.apply(Dataset.scala:1442) 在 org.apache.spark.sql.Dataset$$anonfun$groupBy$2.apply(Dataset.scala:1442) 在 scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala :234) 在 scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala :234) 在 scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala: 59) 在 scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) 在 scala.collection.TraversableLike$class.map(TraversableLike.scala:234) 在 scala.collection.AbstractTraversable.map(Traversable.scala:104) 在 org.apache.spark.sql.Dataset.groupBy(Dataset.scala:1442) ... 58 省略

我还尝试了另一种解决方案。但它为每 60 行提供值而不是聚合。

scala> newVX.groupBy(window($"datetime","1 minute")).agg(avg("value") as "avg-va
lue").show()


17/07/31 12:41:02 WARN Executor: Managed memory leak detected; size = 4456448 by
tes, TID = 5
+--------------------+-------------------+
|              window|          avg-value|
+--------------------+-------------------+
|[2017-07-31 07:49...| 0.7699999809265137|
|[2017-07-31 05:34...|0.33500000834465027|
|[2017-07-31 04:26...|0.23999999463558197|
|[2017-07-30 20:04...| 0.9399999976158142|
|[2017-07-29 08:33...|0.20250000059604645|
|[2017-07-28 09:30...| 0.3400000035762787|
|[2017-07-27 16:36...| 1.2799999713897705|
|[2017-07-27 08:16...| 0.3400000035762787|
|[2017-07-27 08:11...| 0.3400000035762787|
|[2017-07-27 01:06...| 0.4650000035762787|
|[2017-07-26 23:53...|0.23999999463558197|
|[2017-07-26 19:49...| 0.3199999928474426|
|[2017-07-25 14:39...| 0.3400000035762787|
|[2017-07-25 07:54...| 0.7099999785423279|
|[2017-07-25 06:21...|0.29499998688697815|
|[2017-07-25 03:57...| 0.1899999976158142|
|[2017-07-24 20:31...| 1.2799999713897705|
|[2017-07-24 19:50...| 1.2799999713897705|
|[2017-07-24 16:26...|0.03999999910593033|
|[2017-07-24 16:10...|              0.125|
+--------------------+-------------------+
only showing top 20 rows

编辑：我在这里做了一些更正，但仍然显示错误的结果。我通过给出保留日期直到分钟值

val VX = newvx.withColumn("datetime", ((unix_timestamp($"datetime") / 60)
.cast("long") * 60).cast("timestamp"))

在我聚合后它仍然显示错误的值。

scala> VX.groupBy("datetime").agg(Map("value" -> "mean")).show
17/07/31 15:58:15 WARN Executor: Managed memory leak detected; size = 4456448 by
tes, TID = 21
+-------------------+-------------------+
|           datetime|         avg(value)|
+-------------------+-------------------+
|2017-07-31 06:38:00| 0.6100000143051147|
|2017-07-30 19:46:00| 0.3400000035762787|
|2017-07-30 09:24:00|0.42500001192092896|
|2017-07-29 08:53:00| 0.8899999856948853|
|2017-07-29 15:07:00| 0.3400000035762787|
|2017-07-29 05:26:00| 0.3100000023841858|
|2017-07-28 23:29:00|0.27250000834465027|
|2017-07-28 22:07:00| 0.3199999928474426|
|2017-07-28 20:48:00| 0.2849999964237213|
|2017-07-28 20:13:00|0.44999998807907104|
|2017-07-28 18:07:00|0.20999999344348907|
|2017-07-28 06:38:00|0.08500000089406967|
|2017-07-27 11:27:00|0.26499998569488525|
|2017-07-27 02:37:00| 1.0549999475479126|
|2017-07-27 02:12:00| 0.3449999988079071|
|2017-07-26 22:22:00| 0.4699999988079071|
|2017-07-25 15:22:00| 0.8199999928474426|
|2017-07-25 07:08:00| 0.2800000011920929|
|2017-07-25 06:42:00|0.32499998807907104|
|2017-07-25 04:42:00|0.30000001192092896|
+-------------------+-------------------+
only showing top 20 rows

知道为什么吗？我该如何纠正？谢谢。

【问题讨论】：

值怎么不正确？
我将时间设为“1 分钟”。但它没有给出“1分钟”的价值。从上面可以看出，它给出了表格中的随机值。它必须从 2017-07-31 10:53:.. 开始，但输出从 2017-07-31 06:38:00 开始
你在使用show之前有试过排序吗？
已经排好序了。我看到聚合发生在几个小时，而不是每分钟
尝试将您的数据减少到一个可重复性最低的示例 - 可能只有三个日期和值，这样会更明显地发生什么。

标签： scala apache-spark apache-spark-sql aggregate apache-spark-2.0

【解决方案1】：

您可以使用以下方法：

在 spark 中创建一个只保留日期的用户定义函数直到微小的谷物水平。例如 2017-07-31 10:53

def atMinute = udf((dateTime: String) => // implement here retain date till minute)

使用 udf 从初始数据帧转换

val df_at_minute = df.withColumn("datetime_at_min", atMinute("datetime"))

将聚合函数应用于新数据框

 df_at_minute.groupBy("datetime_at_min").agg(avg("value"))

【讨论】：

嗨，我通过这样做将日期保留到分钟：val VX = newvx.withColumn("datetime", ((unix_timestamp($"datetime") / 60) .cast("long") * 60).cast("timestamp"))，它工作得非常好。但是当我groupyBy 并使用avg 时，它仍然给出随机结果。
我认为这是一个与您如何计算日期时间字段更相关的问题，导致聚合它；没关系