ksml/examples/09-example-aggregate.yaml at main · Axual/ksml · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
# $schema: https://axual.github.io/ksml/latest/ksml-language-spec.json

# This example shows how to read from a simple stream, group by owner, apply windows and count owners per window.

streams:
  sensor_source:
    topic: ksml_sensordata_avro
    keyType: string
    valueType: avro:SensorData
    offsetResetPolicy: latest

pipelines:
  main:
    from: sensor_source
    via:
      - type: groupBy
        name: group_by_owner
        mapper:
          code: |
            if value is None or not "owner" in value:
              return "no owner"
          expression: value["owner"]
          resultType: string
      - type: windowByTime
        windowType: tumbling
        duration: 20s
      - type: aggregate
        store:
          type: window
          windowSize: 10m
          retention: 1h
          caching: false
        initializer:
          expression: 0
          resultType: long
        aggregator:
          expression: aggregatedValue+1
          resultType: long
      - type: toStream
      - type: peek
        forEach:
          code: log.info("AGGREGATE - key={}, value={}", key, value)