sdks/python/apache_beam/yaml/tests/windowinto.yaml

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

pipelines:
 - pipeline:
      type: composite
      transforms:
        # setup
        - type: Create
          config:
            elements:
              - {k: "x", t: 1}
              - {k: "x", t: 8}
              - {k: "x", t: 11}
              - {k: "y", t: 101}
        - type: AssignTimestamps
          input: Create
          config:
            timestamp: t

        # global windowing
        - type: chain
          name: Global
          input: AssignTimestamps
          transforms:
            - type: WindowInto
              config:
                windowing:
                  type: global
            - type: Combine
              config:
                group_by: 'k'
                combine:
                  t: sum
            - type: AssertEqual
              config:
                elements:
                  - {k: "x", t: 20}
                  - {k: "y", t: 101}

        # fixed windowing
        - type: chain
          name: Fixed
          input: AssignTimestamps
          transforms:

            - type: WindowInto
              config:
                windowing:
                  type: fixed
                  size: 10s
                  offset: 5s
            - type: Combine
              config:
                group_by: 'k'
                combine:
                  t: sum
            - type: AssertEqual
              config:
                elements:
                  - {k: "x", t: 1}
                  - {k: "x", t: 19}
                  - {k: "y", t: 101}

        # sliding windowing
        - type: chain
          name: Sliding
          input: AssignTimestamps
          transforms:
            - type: WindowInto
              config:
                windowing:
                  type: sliding
                  size: 20s
                  period: 10s
            - type: Combine
              config:
                group_by: 'k'
                combine:
                  t: sum
            - type: AssertEqual
              config:
                elements:
                  - {k: "x", t: 9}    # [-10, 10)
                  - {k: "x", t: 20}   # [  0, 20)
                  - {k: "x", t: 11}   # [ 10, 30)
                  - {k: "y", t: 101}  # [90, 110)
                  - {k: "y", t: 101}  # [100, 120)

        # session windowing
        - type: chain
          name: Sessions
          input: AssignTimestamps
          transforms:
            - type: WindowInto
              config:
                windowing:
                  type: sessions
                  gap: 5s
            - type: Combine
              config:
                group_by: 'k'
                combine:
                  t: sum
            - type: AssertEqual
              config:
                elements:
                  - {k: "x", t: 1}
                  - {k: "x", t: 19}
                  - {k: "y", t: 101}