ℹ️ Select 'Choose Exercise', or randomize 'Next Random Exercise' in selected language.

Choose Exercise:
Timer 00:00
WPM --
Score --
Acc --
Correct chars --

Scala Stream Processing: Aggregating Event Data

Scala

Goal -- WPM

Ready
Exercise Algorithm Area
1package com.example.streams
2
3import scala.collection.immutable.Seq
4
5// Represents a single event with a type and a value.
6case class Event(eventType: String, value: Double)
7
8// Holds the aggregated results for a given event type.
9case class AggregatedData(totalValue: Double, count: Long)
10
11object StreamAggregator {
12
13/**
14* Aggregates event data from a sequence of events.
15*
16* This function processes a stream of `Event` objects and computes
17* the total value and count for each distinct `eventType`.
18*
19* @param events The input stream of `Event` objects.
20* @return A map where keys are event types and values are `AggregatedData`.
21*/
22def aggregateEvents(events: Seq[Event]): Map[String, AggregatedData] = {
23// Handle the edge case of an empty input stream.
24if (events.isEmpty) {
25Map.empty[String, AggregatedData]
26} else {
27// Use foldLeft for efficient aggregation.
28events.foldLeft(Map.empty[String, AggregatedData]) { (acc, event) =>
29// Retrieve existing aggregated data for the current event type, or default to zero.
30val currentData = acc.getOrElse(event.eventType, AggregatedData(0.0, 0L))
31
32// Update the aggregated data.
33val updatedData = AggregatedData(
34totalValue = currentData.totalValue + event.value,
35count = currentData.count + 1
36)
37
38// Insert the updated data back into the accumulator map.
39acc + (event.eventType -> updatedData)
40}
41}
42}
43
44/**
45* A helper function to process a single event and update the accumulator.
46* This is an internal detail of the aggregation process.
47*
48* @param acc The current accumulator map.
49* @param event The event to process.
50* @return The updated accumulator map.
51*/
52private def processEvent(acc: Map[String, AggregatedData], event: Event): Map[String, AggregatedData] = {
53val currentData = acc.getOrElse(event.eventType, AggregatedData(0.0, 0L))
54val updatedData = AggregatedData(
55totalValue = currentData.totalValue + event.value,
56count = currentData.count + 1
57)
58acc + (event.eventType -> updatedData)
59}
60
61// Example usage:
62def main(args: Array[String]): Unit = {
63val sampleEvents = Seq(
64Event("click", 1.0),
65Event("view", 5.0),
66Event("click", 2.0),
67Event("purchase", 100.0),
68Event("view", 3.0),
69Event("click", 1.5)
70)
71
72val aggregated = aggregateEvents(sampleEvents)
73println("Aggregated Data:")
74aggregated.foreach { case (eventType, data) =>
75println(s" $eventType: Total Value = ${data.totalValue}, Count = ${data.count}")
76}
77
78// Test with an empty stream
79val emptyAggregated = aggregateEvents(Seq.empty[Event])
80println("\nAggregated Data for empty stream:")
81println(emptyAggregated)
82}
83}
Algorithm description viewbox

Scala Stream Processing: Aggregating Event Data

Algorithm description:

This Scala code defines a function `aggregateEvents` that processes a sequence of `Event` objects. It aggregates data by calculating the total value and the number of occurrences for each distinct event type. This is useful in analytics platforms for summarizing user interactions or system metrics over time.

Algorithm explanation:

The `aggregateEvents` function utilizes `foldLeft` to iterate through the input `Seq[Event]`. The accumulator is a `Map[String, AggregatedData]`, where the key is the event type and the value holds the running total and count. For each event, it retrieves the current aggregated data for that type (defaulting to zero if not present), updates the total value and count, and then inserts the new aggregated data back into the map. The time complexity is O(N), where N is the number of events, as each event is processed once. The space complexity is O(M), where M is the number of unique event types, to store the aggregated results. Edge cases like an empty input sequence are handled by returning an empty map immediately. The correctness is maintained by the immutable nature of the map updates and the clear aggregation logic within the fold.

Pseudocode:

FUNCTION aggregateEvents(events):
  IF events is empty THEN
    RETURN empty map
  END IF

  Initialize an empty map called 'accumulator'

  FOR EACH event IN events:
    GET current_data for event.eventType from accumulator, default to (totalValue=0, count=0)
    CALCULATE updated_data: totalValue = current_data.totalValue + event.value, count = current_data.count + 1
    UPDATE accumulator with event.eventType mapped to updated_data
  END FOR

  RETURN accumulator
END FUNCTION