-
Notifications
You must be signed in to change notification settings - Fork 4.5k
Expand file tree
/
Copy pathwindowinto.yaml
More file actions
125 lines (118 loc) · 3.56 KB
/
windowinto.yaml
File metadata and controls
125 lines (118 loc) · 3.56 KB
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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
#
# 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}