src/services/windows.py
BeforeProposed
+8−2
1from pyspark.sql import functions as F
2
3
14def aggregate(frame, width):
2 # TODO: group study and epoch windows using DataFrame aggregations.
3 return frame.sparkSession.createDataFrame([], "study string, windowStart long, count long, total long, subjects long")
5 buckets = frame.withColumn("windowStart", (F.floor(F.col("eventSecond") / width) * width).cast("long"))
6 return (buckets.groupBy("study", "windowStart")
7 .agg(F.count("id").alias("count"), F.sum("value").alias("total"),
8 F.countDistinct("subject").alias("subjects"))
9 .orderBy("study", "windowStart"))