8-reduce-complex-pipeline.yaml

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: complex-sliding-sum
spec:
  limits:
    readBatchSize: 50
  vertices:
    - name: in
      scale:
        min: 1
      source:
        http: {}
    - name: atoi
      scale:
        min: 1
      udf:
        container:
          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
          image: quay.io/numaio/numaflow-go/map-even-odd:v0.5.0
    - name: keyed-fixed-sum
      partitions: 2
      udf:
        container:
          # compute the sum
          image: quay.io/numaio/numaflow-go/reduce-sum:v0.5.0
        groupBy:
          window:
            fixed:
              length: 5s
          keyed: true
          storage:
            persistentVolumeClaim:
              volumeSize: 10Gi
              accessMode: ReadWriteOnce
    - name: non-keyed-fixed-sum
      udf:
        container:
          # compute the sum
          image: quay.io/numaio/numaflow-go/reduce-sum:v0.5.0
        groupBy:
          window:
            fixed:
              length: 10s
          keyed: false
          storage:
            persistentVolumeClaim:
              volumeSize: 10Gi
              accessMode: ReadWriteOnce
    - name: non-keyed-sliding-sum
      udf:
        container:
          # compute the sum
          image: quay.io/numaio/numaflow-go/reduce-sum:v0.5.0
        groupBy:
          window:
            sliding:
              length: 60s
              slide: 10s
          keyed: false
          storage:
            persistentVolumeClaim:
              volumeSize: 10Gi
              accessMode: ReadWriteOnce
    - name: sink
      scale:
        min: 1
      sink:
        log: {}
  edges:
    - from: in
      to: atoi
    - from: atoi
      to: keyed-fixed-sum
    - from: keyed-fixed-sum
      to: non-keyed-fixed-sum
    - from: non-keyed-fixed-sum
      to: non-keyed-sliding-sum
    - from: non-keyed-sliding-sum
      to: sink