Process a list of optional sensor readings and select the minimum using stream join with <?.

← Streams and Pipelines · Ref: Q1233

Define a pure function using <? and pass it to 'join with' in a stream pipeline. The join operation reduces the stream to a single value.

  coolerOf() as pure
    -> left as SensorReading, right as SensorReading
    <- rtn as SensorReading: left <? right

  coolest <- cat readings | join with coolerOf | collect as SensorReading

The <? operator handles unset values gracefully within the stream. The custom <=> on temperature determines ordering.

See Q1139 for join basics. See Q1230 for custom <=> on records.

Example

defines module qa.streams.coalesce.pipeline

  defines record

    SensorReading
      stationId as String: String()
      temperature as Float: 0.0

      SensorReading()
        ->
          stationId as String
          temperature as Float
        this.stationId: stationId
        this.temperature: temperature

      operator <=> as pure
        -> other as SensorReading
        <- rtn as Integer: temperature <=> other.temperature

      default operator

  defines function

    coolerOf() as pure
      ->
        left as SensorReading
        right as SensorReading
      <- rtn as SensorReading: left <? right

    warmerOf() as pure
      ->
        left as SensorReading
        right as SensorReading
      <- rtn as SensorReading: left >? right

  defines program

    CoalesceInPipelineDemo()
      stdout <- Stdout()

      // === BUILD A LIST OF SENSOR READINGS ===

      readings <- List() of SensorReading
      readings += SensorReading("North", 22.5)
      readings += SensorReading("South", 18.3)
      readings += SensorReading("East", 25.1)
      readings += SensorReading("West", 15.8)

      // === FIND COOLEST USING STREAM JOIN WITH <? ===

      coolestList <- cat readings | join with coolerOf | collect as List of SensorReading
      stdout.println(`Coolest: ${coolestList}`)

      // === FIND WARMEST USING STREAM JOIN WITH >? ===

      warmestList <- cat readings | join with warmerOf | collect as List of SensorReading
      stdout.println(`Warmest: ${warmestList}`)

      // === SORTED OUTPUT ===

      stdout.println("All readings sorted by temperature:")
      cat readings | sort > stdout

Common mistakes

E50060 — EK9 List has no .stream() method; reduce with a stream pipeline 'cat list | join with fn'. See ek9 -h E50060 for details.

Incorrect:

coolestList <- readings.stream().min()

Correct:

coolestList <- cat readings | join with coolerOf | collect as List of SensorReading

Other ways to ask this

How do I reduce a stream of custom records to find the minimum?
Given sensor readings from multiple stations, find the coolest using a pipeline.
In Java I'd use stream().min(Comparator.comparing(SensorReading::getTemperature)). What does EK9 use?
Migrating from Python where I use min(readings, key=lambda r: r.temperature) — how does EK9 do this?

Coming from another language?

Java: stream().min(Comparator.comparingDouble(SensorReading::getTemperature)). Python: min(readings, key=lambda r: r.temperature). Rust: readings.iter().min_by_key(|r| r.temperature). EK9: cat readings | join with coolerOf | collect as SensorReading.

Keywords: reduce, minimum, coalescing, join, stream, <?, sensor, pipeline, record