Sum the prices of all products over 20 using a stream pipeline.

Filter, map to price, then join with an addition function:

  total <- cat products | filter by isExpensive | map with getPrice | join with addFloats | collect as Float

The join operation reduces a stream to a single value via a binary function. Each pipeline stage uses a named pure function. See Q1139 for join basics. See Q235 for stream operations reference.

Example

#!ek9
defines module qa.streams.sumaggregation

  defines class

    Product
      name as String?
      price as Float?

      default private Product() as pure

      Product() as pure
        ->
          name as String
          price as Float
        this.name :=? name
        this.price :=? price

      name() as pure
        <- rtn as String: String(name)

      price() as pure
        <- rtn as Float: Float(price)

      operator $ as pure
        <- rtn as String: `${name} \$${price}`

      override operator ? as pure
        <- rtn as Boolean: name? and price?

  defines function

    isExpensive() as pure
      -> product as Product
      <- rtn as Boolean?
      expensiveThreshold <- 20.0
      rtn: product.price() > expensiveThreshold

    getPrice() as pure
      -> product as Product
      <- rtn as Float: product.price()

    addFloats() as pure
      ->
        a as Float
        b as Float
      <- rtn as Float: a + b

  defines program

    StreamSumDemo()
      stdout <- Stdout()

      products <- [
        Product("Widget", 15.0),
        Product("Gadget", 45.0),
        Product("Gizmo", 30.0),
        Product("Trinket", 5.0),
        Product("Device", 60.0)
        ]

      // Filter to expensive products, extract price, sum
      total <- cat products | filter by isExpensive | map with getPrice | join with addFloats | collect as Float
      stdout.println(`Total of expensive items: ${total}`)

      // Count expensive items by collecting as list and measuring length
      expensiveList <- cat products | filter by isExpensive | collect as List of Product
      stdout.println(`Number of expensive items: ${length expensiveList}`)

Common mistakes

E50060 — EK9 uses cat/pipe stream syntax, not Java-style .stream() calls. Reduction uses 'join with' and a binary function. See ek9 -h E50060 for details.

Incorrect:

total <- products.stream()

Correct:

total <- cat products | filter by isExpensive | map with getPrice | join with addFloats | collect as Float

E07520 — A function used with 'filter by' must return Boolean; making isExpensive return Float triggers E07520 (must return a Boolean). See ek9 -h E07520 for details.

Incorrect:

      <- rtn as Float?
      rtn: product.price()

Correct:

      <- rtn as Boolean?
      expensiveThreshold <- 20.0
      rtn: product.price() > expensiveThreshold

Other ways to ask this

Write code to filter expensive products and sum their prices in a stream
I have a list of products and need the total price of those costing more than 20
Given a product list, compute the sum of prices above a threshold using a pipeline
In Java I'd use stream().filter().mapToDouble().sum(). Write the EK9 equivalent

Coming from another language?

Java: stream().filter(p -> p.price > 20).mapToDouble(Product::getPrice).sum(). Python: sum(p.price for p in products if p.price > 20). Rust: iter().filter().map().sum(). EK9: cat | filter by | map with | join with | collect as.

Keywords: aggregate, filter, join, map, pipeline, price, reduce, stream, sum, total