apache spark |特定时间帧聚合

ovfsdjhp  于 2021-05-27  发布在  Spark
关注(0)|答案(1)|浏览(396)

我需要一种按周汇总数据集的方法。这是我的数据集

|      date|organization_id|media_package_id|event_uuid |
+----------+---------------+----------------+-----------+
|2016-10-25|              1|              11|     76304d|
|2016-10-25|              1|              11|     e6285b|
|2016-10-22|              2|              21|     16c04d|
|2016-10-22|              2|              21|     17804d|
|2016-10-22|              2|              21|     18904x|
|2016-10-21|              2|              21|     51564q|
|2016-10-07|              4|              98|     12874t|
|2016-10-05|              4|              98|     11234d|
+----------+---------------+----------------+-----------+

假设spark作业每天都在运行,以获得所需的聚合结果。我想在一个星期的基础上,例如以上数据集的汇总结果将是。

|      date|organization_id|media_package_id|      count|
+----------+---------------+----------------+-----------+
|2016-10-24|              1|              11|          2|
|2016-10-17|              2|              21|          4|
|2016-10-03|              4|              98|          2|
+----------+---------------+----------------+-----------+

在这里,如果你看到日期栏,它是采取一周的第一天(我认为这是最好的方式)
我设法做了每日的汇总。我是这样做的

val data = MongoSupport.load(spark, "sampleCollection")
val dataForDates = data.filter(dataForDates("date").isin(dates : _*))

val countByDate = proofEventsForDates.groupBy("DATE", "ORGANIZATION_ID", "MEDIA_PACKAGE_ID")
  .agg(count("EVENT_UUID").as("COUNT"))

val finalResult = impressionsByDate
  .select(
    col("DATE").as("date"),
    col("ORGANIZATION_ID").as("organization_id"),
    col("MEDIA_PACKAGE_ID").as("media_package_id"),
    col("COUNT").as("count")
  )

在这里,在开始过滤数据集时,我传递了一个特殊的 dates 包含至少一个月的日期的列表。我得到的结果是(这不是我想要的)

|      date|organization_id|media_package_id|      count|
+----------+---------------+----------------+-----------+
|2016-10-25|              1|              11|          2|
|2016-10-22|              2|              21|          3|
|2016-10-21|              2|              21|          1|
|2016-10-07|              2|              21|          1|
|2016-10-05|              2|              21|          1|
+----------+---------------+----------------+-----------+

从这里开始,我不知道如何每周汇总这个数据集。

ghhkc1vu

ghhkc1vu1#

假设你 date 列已属于类 date ,您可以使用 year() 以及 weekofyear() 提取聚合缺少的分组列。

import org.apache.spark.sql.functions.weekofyear
import org.apache.spark.sql.functions.year

(df
  .withColumn("week_nr", weekofyear($"date"))
  .withColumn("year", year($"date"))
  .groupBy("year",
           "week_nr",
           "organization_id",
           "media_package_id")
  .count().orderBy(desc("week_nr"))).show
+----+-------+---------------+----------------+-----+
|year|week_nr|organization_id|media_package_id|count|
+----+-------+---------------+----------------+-----+
|2016|     43|              1|              11|    2|
|2016|     42|              2|              21|    4|
|2016|     40|              4|              98|    2|
+----+-------+---------------+----------------+-----+

相关问题