Spark Structured Streaming - How to deduplicate by latest and aggregate count

user_2628781 0
Viewed 1604

I would like to perform structured streaming aggregation with a windowing period. Given this following data schema. The objective is to filter by the latest occurring event based on user. Then aggregate the count of each event type for each location.

time    location   user   type
 1        A         1      one
 2        A         1      two
 1        B         2      one
 2        B         2      one
 1        A         3      two
 1        A         4      one

Sample output:

location   countOne   countTwo
    A          1         2
    B          1         0

something like the following:

val aggTypes = df
  .select($"location", $"time", $"user", $"type")
  .groupBy($"user")
  .agg(max($"timestamp") as 'timestamp)
  .select("*")
  .withWatermark("timestamp", conf.kafka.watermark.toString + " seconds")
  .groupBy(functions.window($"timestamp", DataConstant.t15min.toString + " seconds", DataConstant.t1min.toString + " seconds", $"location")
  .agg(count(when($"type" === "one", $"type")) as 'countOne, count(when($"type" === "two", $"type" as 'countTwo)))
  .drop($"window")

As structured streaming does not support multiple aggregations and Non-time-based windows are not supported on streaming DataFrames/Datasets. I am not sure if it is possible to achieve the desired output in 1 streaming query.

Any help is appreciated.

spark-structured-streaming apache-spark-sql apache-spark
1 Answers

Seems like you are trying to do Stateless Aggregations. https://spark.apache.org/docs/2.0.2/api/java/org/apache/spark/sql/KeyValueGroupedDataset.html#flatMapGroups(org.apache.spark.api.java.function.FlatMapGroupsFunction,%20org.apache.spark.sql.Encoder)

flatMapGroups is an aggregation API which applies a function to each group in the dataset. It’s only available on a grouped dataset.flatMapGroups doesn’t support the partial aggregations which increase shuffle overhead. So use this API only to do small batch aggregations which will fit into memory. It is also suggested to use the reduce function or an Aggregator. https://spark.apache.org/docs/2.0.2/api/java/org/apache/spark/sql/expressions/Aggregator.html

val count = words.groupByKey(x => x)
            .flatMapGroups
             {
              case (x, iterator) ⇒ Iterator((x, iterator.length))
              }.toDF("x", "count")        


count.writeStream.format("console").outputMode(OutputMode.Append())
user_78510300
Related