How do I group items and then flatten them back in an EK9 stream?

Group collects items into sub-lists by a key. Flatten expands those sub-lists back into individual items.

IMPORTANT: sort BEFORE group. Items must be sorted by the grouping key before grouping. The compiler enforces this with E10040.

PATTERN:

  cat items | sort by keyComparator | group by keyExtractor | flatten > stdout

STEP BY STEP:

1. sort by — orders items so equal keys are adjacent
2. group by — collects adjacent equal-key items into Lists
3. (optional processing on groups — filter with, map by)
4. flatten — expands each List back to individual items

Without flatten, you get a stream of Lists. With flatten, you get a stream of the original item type.

Example

defines module qa.streams.groupflatten

  defines record

    Employee
      employeeName as String: String()
      department as String: String()
      salary as Float: 0.0

      Employee()
        ->
          employeeName as String
          department as String
          salary as Float
        this.employeeName :=: employeeName
        this.department :=: department
        this.salary :=: salary

      default operator

  defines function

    compareDepartment() as pure
      ->
        left as Employee
        right as Employee
      <- rtn as Integer: left.department <=> right.department

    extractDepartment() as pure
      -> employee as Employee
      <- rtn as String: employee.department

  defines program

    GroupFlattenDemo()
      stdout <- Stdout()

      employees <- [
        Employee("Alice", "Engineering", 95000.0),
        Employee("Bob", "Marketing", 72000.0),
        Employee("Charlie", "Engineering", 88000.0),
        Employee("Diana", "Marketing", 81000.0),
        Employee("Eve", "Engineering", 102000.0)
      ]

      //Sort by department, group, then flatten back
      stdout.println("Employees grouped by department:")
      cat employees
        | sort by compareDepartment
        | group by extractDepartment
        | flatten
        > stdout

Common mistakes

E10040 — EK9 requires sort before group. The grouping algorithm needs sorted input to collect adjacent equal-key items. Add sort by before group by. See ek9 -h E10040.

Incorrect:

      cat employees
        | group by extractDepartment

Correct:

      cat employees
        | sort by compareDepartment
        | group by extractDepartment

Other ways to ask this

Show me group by and flatten in an EK9 stream pipeline
How do I process grouped data then flatten results in EK9?
What does sort by then group by then flatten do in a pipeline?

Coming from another language?

EK9 requires sort before group — the grouping algorithm assumes sorted input. This is enforced at compile time, unlike Java's Collectors.groupingBy which sorts internally.

Keywords: pipeline, group, flatten, sort, stream