scala spark获取每个时间间隔的平均值答案

【问题标题】：scala spark get the mean of each time intervalscala spark获取每个时间间隔的平均值
【发布时间】：2018-10-16 09:40:50
【问题描述】：

我的输入是 Spark 数据框：

EventTime,Signal
0,-65
10,-63
20,-71
40,-65
50,-62
80,-81
90,-84
100,-81
...
85460,-71
85480,-66
85490,-89
85500,-80

我想得到EventTime 每 900 秒的 Signal 平均值，输出如下所示：

EventTime, MeanSignal
0, mean 
900, mean 
1800, mean
...
85500, mean

我的问题是常规数据中没有EventTime的常规步骤，所以我不能将数据帧分成相同长度的部分......

【问题讨论】：

请阅读How to make good reproducible Apache Spark Dataframe examples和edit您的问题。
How to group by time interval in Spark SQL的可能重复
另外PySpark Numeric Window Group By 可能有用

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

【解决方案1】：

您可以添加一个新列作为 EventTime/900 并基于该列进行分组。像这样的。

val map = Map(0 -> -65 , 10 -> -63 , 20 -> -71 , 40 -> -65 , 50 -> -62 , 80 -> -81 , 90 -> -84 , 100 -> -81 , 85460 -> -71 , 85480 -> -66 , 85490 -> -89 , 85500 -> -80)

val df = map.toSeq.toDF("EventTime", "Signal")
  .withColumn("EventTimeGrp", (col("EventTime")/lit(900)).cast("int"))
  .groupBy("EventTimeGrp").agg(mean("Signal").as("MeanSignal"))
  .withColumn("EventTime", col("EventTimeGrp")*lit(900))
  .drop("EventTimeGrp")

结果如下所示。 EventTime 0 代表 0-899 之间的值，以此类推。

+------------------+---------+ | MeanSignal|EventTime| +------------------+---------+ |-75.33333333333333| 84600| | -71.5| 0| | -80.0| 85500| +------------------+---------+

【讨论】：

【解决方案2】：

好的，这是我的解决方案，感谢其他帖子：我创建了一个与 EventTime 的模相关联的列 Bucket 来创建类别，然后我对桶进行分组并取平均值

    val df = data_input.withColumn("Bucket", toBucketUDF(col("EventTime")))

    val finalDF = df.groupBy("Bucket")
      .agg(mean("RSSI"))
      .withColumnRenamed("avg(RSSI)", "RSSI")
      .orderBy("Bucket")
      .withColumn("EventTime", getTimeUDF(col("Bucket")))
      .drop("Bucket")



    finalDF

  }

  def toBucket(input:Int): Int = {
    val Bucket = input/900
    return Bucket
  }
  def getTime(input: Int): Int = {
    val time = (input+1) * 900
    return time
  }

  val toBucketUDF = udf(toBucket _)
  val getTimeUDF = udf(getTime _)

【讨论】：