You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2020/02/08 02:26:08 UTC
[beam] branch master updated: [BEAM-3595] Migrate to "v1" URNs for
standard window fns.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new bd23d08 [BEAM-3595] Migrate to "v1" URNs for standard window fns.
new bde3031 Merge pull request #10790 from lukecwik/beam3595
bd23d08 is described below
commit bd23d083290023621c627022fc039f1fdb9c3234
Author: Luke Cwik <lc...@google.com>
AuthorDate: Thu Feb 6 12:00:11 2020 -0800
[BEAM-3595] Migrate to "v1" URNs for standard window fns.
Add missing documentation.
---
.../pipeline/src/main/proto/beam_runner_api.proto | 2 +-
.../src/main/proto/standard_window_fns.proto | 62 ++++++--
.../construction/WindowingStrategyTranslation.java | 18 ++-
sdks/go/pkg/beam/core/runtime/exec/translate.go | 2 +-
sdks/go/pkg/beam/core/runtime/graphx/translate.go | 10 +-
.../beam/model/pipeline_v1/beam_runner_api.pb.go | 2 +-
.../model/pipeline_v1/standard_window_fns.pb.go | 166 +++++++++++++--------
.../harness/control/ProcessBundleHandlerTest.java | 2 +-
sdks/python/apache_beam/portability/common_urns.py | 4 +-
sdks/python/apache_beam/portability/python_urns.py | 2 +-
sdks/python/apache_beam/transforms/window.py | 5 +-
11 files changed, 180 insertions(+), 95 deletions(-)
diff --git a/model/pipeline/src/main/proto/beam_runner_api.proto b/model/pipeline/src/main/proto/beam_runner_api.proto
index 86aa677..5439bc0 100644
--- a/model/pipeline/src/main/proto/beam_runner_api.proto
+++ b/model/pipeline/src/main/proto/beam_runner_api.proto
@@ -1186,7 +1186,7 @@ extend google.protobuf.EnumValueOptions {
//
// 1. The runner understands the URN. For example, it might be
// a well-known URN like "beam:transform:Top" or
-// "beam:windowfn:FixedWindows" with
+// "beam:window_fn:FixedWindows" with
// an agreed-upon payload (e.g. a number or duration,
// respectively).
// 2. The runner does not understand the URN. It might be an
diff --git a/model/pipeline/src/main/proto/standard_window_fns.proto b/model/pipeline/src/main/proto/standard_window_fns.proto
index 6de9984..da4d53b 100644
--- a/model/pipeline/src/main/proto/standard_window_fns.proto
+++ b/model/pipeline/src/main/proto/standard_window_fns.proto
@@ -33,37 +33,81 @@ import "beam_runner_api.proto";
import "google/protobuf/duration.proto";
import "google/protobuf/timestamp.proto";
+// By default, all data in a PCollection is assigned to the single global
+// window. See BeamConstants for the time span this window encompasses.
+//
+// See https://beam.apache.org/documentation/programming-guide/#single-global-window
+// for additional details.
message GlobalWindowsPayload {
enum Enum {
- // TODO(BEAM-3595): Change this to beam:windowfn:global_windows:v1
- PROPERTIES = 0 [(beam_urn) = "beam:windowfn:global_windows:v0.1"];
+ PROPERTIES = 0 [(beam_urn) = "beam:window_fn:global_windows:v1"];
}
// Empty payload
}
+// A fixed time window represents a consistent duration size, non overlapping
+// time interval in the data stream.
+//
+// See https://beam.apache.org/documentation/programming-guide/#fixed-time-windows
+// for additional details.
message FixedWindowsPayload {
enum Enum {
- // TODO(BEAM-3595): Change this to beam:windowfn:fixed_windows:v1
- PROPERTIES = 0 [(beam_urn) = "beam:windowfn:fixed_windows:v0.1"];
+ PROPERTIES = 0 [(beam_urn) = "beam:window_fn:fixed_windows:v1"];
}
+
+ // (Required) Represents the size of the window.
google.protobuf.Duration size = 1;
+
+ // (Required) Represents the timestamp of when the first window begins.
+ // Window N will start at offset + N * size.
google.protobuf.Timestamp offset = 2;
}
+// A sliding time window represents time intervals in the data stream that can
+// overlap. For example, each window might capture 60 seconds worth of data, but
+// a new window starts every 30 seconds. The frequency with which sliding
+// windows begin is called the period. Therefore, our example would have a
+// window size of 60 seconds and a period of 30 seconds.
+//
+// Because multiple windows overlap, most elements in a data set will belong to
+// more than one window. This kind of windowing is useful for taking running
+// averages of data; using sliding time windows, you can compute a running
+// average of the past 60 seconds’ worth of data, updated every 30 seconds, in
+// our example.
+//
+// See https://beam.apache.org/documentation/programming-guide/#sliding-time-windows
+// for additional details.
message SlidingWindowsPayload {
enum Enum {
- // TODO(BEAM-3595): Change this to beam:windowfn:sliding_windows:v1
- PROPERTIES = 0 [(beam_urn) = "beam:windowfn:sliding_windows:v0.1"];
+ PROPERTIES = 0 [(beam_urn) = "beam:window_fn:sliding_windows:v1"];
}
+
+ // (Required) Represents the size of the window.
google.protobuf.Duration size = 1;
+
+ // (Required) Represents the timestamp of when the first window begins.
+ // Window N will start at offset + N * period.
google.protobuf.Timestamp offset = 2;
+
+ // (Required) Represents the amount of time between each start of a window.
google.protobuf.Duration period = 3;
}
-message SessionsPayload {
+// A session window function defines windows that contain elements that are
+// within a certain gap size of another element. Session windowing applies
+// on a per-key basis and is useful for data that is irregularly distributed
+// with respect to time. For example, a data stream representing user mouse
+// activity may have long periods of idle time interspersed with high
+// concentrations of clicks. If data arrives after the minimum specified gap
+// size duration, this initiates the start of a new window.
+//
+// See https://beam.apache.org/documentation/programming-guide/#session-windows
+// for additional details.
+message SessionWindowsPayload {
enum Enum {
- // TODO(BEAM-3595): Change this to beam:windowfn:session_windows:v1
- PROPERTIES = 0 [(beam_urn) = "beam:windowfn:session_windows:v0.1"];
+ PROPERTIES = 0 [(beam_urn) = "beam:window_fn:session_windows:v1"];
}
+
+ // (Required) Minimum duration of gaps between sessions.
google.protobuf.Duration gap_size = 1;
}
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
index bbb31a7..f8a8648 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
@@ -27,7 +27,7 @@ import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec;
import org.apache.beam.model.pipeline.v1.RunnerApi.OutputTime;
import org.apache.beam.model.pipeline.v1.StandardWindowFns.FixedWindowsPayload;
import org.apache.beam.model.pipeline.v1.StandardWindowFns.GlobalWindowsPayload;
-import org.apache.beam.model.pipeline.v1.StandardWindowFns.SessionsPayload;
+import org.apache.beam.model.pipeline.v1.StandardWindowFns.SessionWindowsPayload;
import org.apache.beam.model.pipeline.v1.StandardWindowFns.SlidingWindowsPayload;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
@@ -201,14 +201,15 @@ public class WindowingStrategyTranslation implements Serializable {
// This URN says that the WindowFn is just a UDF blob the Java SDK understands
// TODO: standardize such things
- public static final String SERIALIZED_JAVA_WINDOWFN_URN = "beam:windowfn:javasdk:v0.1";
+ public static final String SERIALIZED_JAVA_WINDOWFN_URN = "beam:window_fn:serialized_java:v1";
public static final String GLOBAL_WINDOWS_URN =
BeamUrns.getUrn(GlobalWindowsPayload.Enum.PROPERTIES);
public static final String FIXED_WINDOWS_URN =
BeamUrns.getUrn(FixedWindowsPayload.Enum.PROPERTIES);
public static final String SLIDING_WINDOWS_URN =
BeamUrns.getUrn(SlidingWindowsPayload.Enum.PROPERTIES);
- public static final String SESSION_WINDOWS_URN = BeamUrns.getUrn(SessionsPayload.Enum.PROPERTIES);
+ public static final String SESSION_WINDOWS_URN =
+ BeamUrns.getUrn(SessionWindowsPayload.Enum.PROPERTIES);
/**
* Converts a {@link WindowFn} into a {@link RunnerApi.MessageWithComponents} where {@link
@@ -241,13 +242,13 @@ public class WindowingStrategyTranslation implements Serializable {
.setPayload(slidingWindowsPayload.toByteString())
.build();
} else if (windowFn instanceof Sessions) {
- SessionsPayload sessionsPayload =
- SessionsPayload.newBuilder()
+ SessionWindowsPayload sessionWindowsPayload =
+ SessionWindowsPayload.newBuilder()
.setGapSize(Durations.fromMillis(((Sessions) windowFn).getGapDuration().getMillis()))
.build();
return FunctionSpec.newBuilder()
.setUrn(SESSION_WINDOWS_URN)
- .setPayload(sessionsPayload.toByteString())
+ .setPayload(sessionWindowsPayload.toByteString())
.build();
} else {
return FunctionSpec.newBuilder()
@@ -358,8 +359,9 @@ public class WindowingStrategyTranslation implements Serializable {
return SlidingWindows.of(Duration.millis(Durations.toMillis(slidingParams.getSize())))
.every(Duration.millis(Durations.toMillis(slidingParams.getPeriod())))
.withOffset(Duration.millis(Timestamps.toMillis(slidingParams.getOffset())));
- } else if (s.equals(getUrn(SessionsPayload.Enum.PROPERTIES))) {
- SessionsPayload sessionParams = SessionsPayload.parseFrom(windowFnSpec.getPayload());
+ } else if (s.equals(getUrn(SessionWindowsPayload.Enum.PROPERTIES))) {
+ SessionWindowsPayload sessionParams =
+ SessionWindowsPayload.parseFrom(windowFnSpec.getPayload());
return Sessions.withGapDuration(
Duration.millis(Durations.toMillis(sessionParams.getGapSize())));
} else if (s.equals(SERIALIZED_JAVA_WINDOWFN_URN)) {
diff --git a/sdks/go/pkg/beam/core/runtime/exec/translate.go b/sdks/go/pkg/beam/core/runtime/exec/translate.go
index ec57471..fe9eabd 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/translate.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/translate.go
@@ -200,7 +200,7 @@ func unmarshalWindowFn(wfn *pb.FunctionSpec) (*window.Fn, error) {
return window.NewSlidingWindows(period, size), nil
case graphx.URNSessionsWindowFn:
- var payload pb.SessionsPayload
+ var payload pb.SessionWindowsPayload
if err := proto.Unmarshal(wfn.GetPayload(), &payload); err != nil {
return nil, err
}
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/translate.go b/sdks/go/pkg/beam/core/runtime/graphx/translate.go
index 2f915c5..42e0ef3 100644
--- a/sdks/go/pkg/beam/core/runtime/graphx/translate.go
+++ b/sdks/go/pkg/beam/core/runtime/graphx/translate.go
@@ -43,10 +43,10 @@ const (
// URNIterableSideInput = "beam:side_input:iterable:v1"
URNMultimapSideInput = "beam:side_input:multimap:v1"
- URNGlobalWindowsWindowFn = "beam:windowfn:global_windows:v0.1"
- URNFixedWindowsWindowFn = "beam:windowfn:fixed_windows:v0.1"
- URNSlidingWindowsWindowFn = "beam:windowfn:sliding_windows:v0.1"
- URNSessionsWindowFn = "beam:windowfn:session_windows:v0.1"
+ URNGlobalWindowsWindowFn = "beam:window_fn:global_windows:v1"
+ URNFixedWindowsWindowFn = "beam:window_fn:fixed_windows:v1"
+ URNSlidingWindowsWindowFn = "beam:window_fn:sliding_windows:v1"
+ URNSessionsWindowFn = "beam:window_fn:session_windows:v1"
// SDK constants
@@ -529,7 +529,7 @@ func makeWindowFn(w *window.Fn) *pb.FunctionSpec {
return &pb.FunctionSpec{
Urn: URNSessionsWindowFn,
Payload: protox.MustEncode(
- &pb.SessionsPayload{
+ &pb.SessionWindowsPayload{
GapSize: ptypes.DurationProto(w.Gap),
},
),
diff --git a/sdks/go/pkg/beam/model/pipeline_v1/beam_runner_api.pb.go b/sdks/go/pkg/beam/model/pipeline_v1/beam_runner_api.pb.go
index 41873a1..95ed569 100644
--- a/sdks/go/pkg/beam/model/pipeline_v1/beam_runner_api.pb.go
+++ b/sdks/go/pkg/beam/model/pipeline_v1/beam_runner_api.pb.go
@@ -4270,7 +4270,7 @@ func (m *ExternalPayload) GetParams() map[string]string {
//
// 1. The runner understands the URN. For example, it might be
// a well-known URN like "beam:transform:Top" or
-// "beam:windowfn:FixedWindows" with
+// "beam:window_fn:FixedWindows" with
// an agreed-upon payload (e.g. a number or duration,
// respectively).
// 2. The runner does not understand the URN. It might be an
diff --git a/sdks/go/pkg/beam/model/pipeline_v1/standard_window_fns.pb.go b/sdks/go/pkg/beam/model/pipeline_v1/standard_window_fns.pb.go
index 0dcdba8..58a2a8b 100644
--- a/sdks/go/pkg/beam/model/pipeline_v1/standard_window_fns.pb.go
+++ b/sdks/go/pkg/beam/model/pipeline_v1/standard_window_fns.pb.go
@@ -25,7 +25,6 @@ const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
type GlobalWindowsPayload_Enum int32
const (
- // TODO(BEAM-3595): Change this to beam:windowfn:global_windows:v1
GlobalWindowsPayload_PROPERTIES GlobalWindowsPayload_Enum = 0
)
@@ -48,7 +47,6 @@ func (GlobalWindowsPayload_Enum) EnumDescriptor() ([]byte, []int) {
type FixedWindowsPayload_Enum int32
const (
- // TODO(BEAM-3595): Change this to beam:windowfn:fixed_windows:v1
FixedWindowsPayload_PROPERTIES FixedWindowsPayload_Enum = 0
)
@@ -71,7 +69,6 @@ func (FixedWindowsPayload_Enum) EnumDescriptor() ([]byte, []int) {
type SlidingWindowsPayload_Enum int32
const (
- // TODO(BEAM-3595): Change this to beam:windowfn:sliding_windows:v1
SlidingWindowsPayload_PROPERTIES SlidingWindowsPayload_Enum = 0
)
@@ -91,29 +88,33 @@ func (SlidingWindowsPayload_Enum) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_fab9dd76b0d0d680, []int{2, 0}
}
-type SessionsPayload_Enum int32
+type SessionWindowsPayload_Enum int32
const (
- // TODO(BEAM-3595): Change this to beam:windowfn:session_windows:v1
- SessionsPayload_PROPERTIES SessionsPayload_Enum = 0
+ SessionWindowsPayload_PROPERTIES SessionWindowsPayload_Enum = 0
)
-var SessionsPayload_Enum_name = map[int32]string{
+var SessionWindowsPayload_Enum_name = map[int32]string{
0: "PROPERTIES",
}
-var SessionsPayload_Enum_value = map[string]int32{
+var SessionWindowsPayload_Enum_value = map[string]int32{
"PROPERTIES": 0,
}
-func (x SessionsPayload_Enum) String() string {
- return proto.EnumName(SessionsPayload_Enum_name, int32(x))
+func (x SessionWindowsPayload_Enum) String() string {
+ return proto.EnumName(SessionWindowsPayload_Enum_name, int32(x))
}
-func (SessionsPayload_Enum) EnumDescriptor() ([]byte, []int) {
+func (SessionWindowsPayload_Enum) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_fab9dd76b0d0d680, []int{3, 0}
}
+// By default, all data in a PCollection is assigned to the single global
+// window. See BeamConstants for the time span this window encompasses.
+//
+// See https://beam.apache.org/documentation/programming-guide/#single-global-window
+// for additional details.
type GlobalWindowsPayload struct {
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
@@ -145,8 +146,16 @@ func (m *GlobalWindowsPayload) XXX_DiscardUnknown() {
var xxx_messageInfo_GlobalWindowsPayload proto.InternalMessageInfo
+// A fixed time window represents a consistent duration size, non overlapping
+// time interval in the data stream.
+//
+// See https://beam.apache.org/documentation/programming-guide/#fixed-time-windows
+// for additional details.
type FixedWindowsPayload struct {
- Size *duration.Duration `protobuf:"bytes,1,opt,name=size,proto3" json:"size,omitempty"`
+ // (Required) Represents the size of the window.
+ Size *duration.Duration `protobuf:"bytes,1,opt,name=size,proto3" json:"size,omitempty"`
+ // (Required) Represents the timestamp of when the first window begins.
+ // Window N will start at offset + N * size.
Offset *timestamp.Timestamp `protobuf:"bytes,2,opt,name=offset,proto3" json:"offset,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
@@ -192,13 +201,31 @@ func (m *FixedWindowsPayload) GetOffset() *timestamp.Timestamp {
return nil
}
+// A sliding time window represents time intervals in the data stream that can
+// overlap. For example, each window might capture 60 seconds worth of data, but
+// a new window starts every 30 seconds. The frequency with which sliding
+// windows begin is called the period. Therefore, our example would have a
+// window size of 60 seconds and a period of 30 seconds.
+//
+// Because multiple windows overlap, most elements in a data set will belong to
+// more than one window. This kind of windowing is useful for taking running
+// averages of data; using sliding time windows, you can compute a running
+// average of the past 60 seconds’ worth of data, updated every 30 seconds, in
+// our example.
+//
+// See https://beam.apache.org/documentation/programming-guide/#sliding-time-windows
+// for additional details.
type SlidingWindowsPayload struct {
- Size *duration.Duration `protobuf:"bytes,1,opt,name=size,proto3" json:"size,omitempty"`
- Offset *timestamp.Timestamp `protobuf:"bytes,2,opt,name=offset,proto3" json:"offset,omitempty"`
- Period *duration.Duration `protobuf:"bytes,3,opt,name=period,proto3" json:"period,omitempty"`
- XXX_NoUnkeyedLiteral struct{} `json:"-"`
- XXX_unrecognized []byte `json:"-"`
- XXX_sizecache int32 `json:"-"`
+ // (Required) Represents the size of the window.
+ Size *duration.Duration `protobuf:"bytes,1,opt,name=size,proto3" json:"size,omitempty"`
+ // (Required) Represents the timestamp of when the first window begins.
+ // Window N will start at offset + N * period.
+ Offset *timestamp.Timestamp `protobuf:"bytes,2,opt,name=offset,proto3" json:"offset,omitempty"`
+ // (Required) Represents the amount of time between each start of a window.
+ Period *duration.Duration `protobuf:"bytes,3,opt,name=period,proto3" json:"period,omitempty"`
+ XXX_NoUnkeyedLiteral struct{} `json:"-"`
+ XXX_unrecognized []byte `json:"-"`
+ XXX_sizecache int32 `json:"-"`
}
func (m *SlidingWindowsPayload) Reset() { *m = SlidingWindowsPayload{} }
@@ -247,39 +274,50 @@ func (m *SlidingWindowsPayload) GetPeriod() *duration.Duration {
return nil
}
-type SessionsPayload struct {
+// A session window function defines windows that contain elements that are
+// within a certain gap size of another element. Session windowing applies
+// on a per-key basis and is useful for data that is irregularly distributed
+// with respect to time. For example, a data stream representing user mouse
+// activity may have long periods of idle time interspersed with high
+// concentrations of clicks. If data arrives after the minimum specified gap
+// size duration, this initiates the start of a new window.
+//
+// See https://beam.apache.org/documentation/programming-guide/#session-windows
+// for additional details.
+type SessionWindowsPayload struct {
+ // (Required) Minimum duration of gaps between sessions.
GapSize *duration.Duration `protobuf:"bytes,1,opt,name=gap_size,json=gapSize,proto3" json:"gap_size,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
-func (m *SessionsPayload) Reset() { *m = SessionsPayload{} }
-func (m *SessionsPayload) String() string { return proto.CompactTextString(m) }
-func (*SessionsPayload) ProtoMessage() {}
-func (*SessionsPayload) Descriptor() ([]byte, []int) {
+func (m *SessionWindowsPayload) Reset() { *m = SessionWindowsPayload{} }
+func (m *SessionWindowsPayload) String() string { return proto.CompactTextString(m) }
+func (*SessionWindowsPayload) ProtoMessage() {}
+func (*SessionWindowsPayload) Descriptor() ([]byte, []int) {
return fileDescriptor_fab9dd76b0d0d680, []int{3}
}
-func (m *SessionsPayload) XXX_Unmarshal(b []byte) error {
- return xxx_messageInfo_SessionsPayload.Unmarshal(m, b)
+func (m *SessionWindowsPayload) XXX_Unmarshal(b []byte) error {
+ return xxx_messageInfo_SessionWindowsPayload.Unmarshal(m, b)
}
-func (m *SessionsPayload) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
- return xxx_messageInfo_SessionsPayload.Marshal(b, m, deterministic)
+func (m *SessionWindowsPayload) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+ return xxx_messageInfo_SessionWindowsPayload.Marshal(b, m, deterministic)
}
-func (m *SessionsPayload) XXX_Merge(src proto.Message) {
- xxx_messageInfo_SessionsPayload.Merge(m, src)
+func (m *SessionWindowsPayload) XXX_Merge(src proto.Message) {
+ xxx_messageInfo_SessionWindowsPayload.Merge(m, src)
}
-func (m *SessionsPayload) XXX_Size() int {
- return xxx_messageInfo_SessionsPayload.Size(m)
+func (m *SessionWindowsPayload) XXX_Size() int {
+ return xxx_messageInfo_SessionWindowsPayload.Size(m)
}
-func (m *SessionsPayload) XXX_DiscardUnknown() {
- xxx_messageInfo_SessionsPayload.DiscardUnknown(m)
+func (m *SessionWindowsPayload) XXX_DiscardUnknown() {
+ xxx_messageInfo_SessionWindowsPayload.DiscardUnknown(m)
}
-var xxx_messageInfo_SessionsPayload proto.InternalMessageInfo
+var xxx_messageInfo_SessionWindowsPayload proto.InternalMessageInfo
-func (m *SessionsPayload) GetGapSize() *duration.Duration {
+func (m *SessionWindowsPayload) GetGapSize() *duration.Duration {
if m != nil {
return m.GapSize
}
@@ -290,41 +328,41 @@ func init() {
proto.RegisterEnum("org.apache.beam.model.pipeline.v1.GlobalWindowsPayload_Enum", GlobalWindowsPayload_Enum_name, GlobalWindowsPayload_Enum_value)
proto.RegisterEnum("org.apache.beam.model.pipeline.v1.FixedWindowsPayload_Enum", FixedWindowsPayload_Enum_name, FixedWindowsPayload_Enum_value)
proto.RegisterEnum("org.apache.beam.model.pipeline.v1.SlidingWindowsPayload_Enum", SlidingWindowsPayload_Enum_name, SlidingWindowsPayload_Enum_value)
- proto.RegisterEnum("org.apache.beam.model.pipeline.v1.SessionsPayload_Enum", SessionsPayload_Enum_name, SessionsPayload_Enum_value)
+ proto.RegisterEnum("org.apache.beam.model.pipeline.v1.SessionWindowsPayload_Enum", SessionWindowsPayload_Enum_name, SessionWindowsPayload_Enum_value)
proto.RegisterType((*GlobalWindowsPayload)(nil), "org.apache.beam.model.pipeline.v1.GlobalWindowsPayload")
proto.RegisterType((*FixedWindowsPayload)(nil), "org.apache.beam.model.pipeline.v1.FixedWindowsPayload")
proto.RegisterType((*SlidingWindowsPayload)(nil), "org.apache.beam.model.pipeline.v1.SlidingWindowsPayload")
- proto.RegisterType((*SessionsPayload)(nil), "org.apache.beam.model.pipeline.v1.SessionsPayload")
+ proto.RegisterType((*SessionWindowsPayload)(nil), "org.apache.beam.model.pipeline.v1.SessionWindowsPayload")
}
func init() { proto.RegisterFile("standard_window_fns.proto", fileDescriptor_fab9dd76b0d0d680) }
var fileDescriptor_fab9dd76b0d0d680 = []byte{
- // 407 bytes of a gzipped FileDescriptorProto
- 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xbc, 0x93, 0x31, 0x4f, 0xdb, 0x40,
- 0x14, 0xc7, 0xeb, 0x36, 0x4d, 0xab, 0xcb, 0xd0, 0xd6, 0x6d, 0xa4, 0xc4, 0x43, 0x9b, 0x78, 0x68,
- 0xb3, 0xf4, 0x52, 0xa7, 0x55, 0x41, 0x19, 0x00, 0x05, 0x12, 0xc4, 0x44, 0x64, 0x47, 0x8a, 0xc4,
- 0x62, 0x9d, 0xb9, 0xb3, 0x39, 0xc9, 0xbe, 0x3b, 0xf9, 0xec, 0x04, 0xf2, 0x0d, 0xf8, 0x1a, 0x7c,
- 0x06, 0x06, 0x66, 0xbe, 0x10, 0x3b, 0x13, 0xca, 0xf9, 0x82, 0xe4, 0x30, 0x04, 0x16, 0x46, 0xfb,
- 0xfd, 0xdf, 0x7b, 0xbf, 0x9f, 0x4e, 0x0f, 0x34, 0x65, 0x86, 0x18, 0x46, 0x29, 0xf6, 0xe7, 0x94,
- 0x61, 0x3e, 0xf7, 0x43, 0x26, 0xa1, 0x48, 0x79, 0xc6, 0xcd, 0x36, 0x4f, 0x23, 0x88, 0x04, 0x3a,
- 0x3d, 0x23, 0x30, 0x20, 0x28, 0x81, 0x09, 0xc7, 0x24, 0x86, 0x82, 0x0a, 0x12, 0x53, 0x46, 0xe0,
- 0xcc, 0xb1, 0xea, 0xcb, 0xff, 0x7e, 0x9a, 0x33, 0x46, 0x52, 0x1f, 0x09, 0x5a, 0x74, 0x5a, 0xdf,
- 0x23, 0xce, 0xa3, 0x98, 0x74, 0xd5, 0x57, 0x90, 0x87, 0x5d, 0x9c, 0xa7, 0x28, 0xa3, 0x9c, 0xe9,
- 0xfa, 0x8f, 0xf5, 0x7a, 0x46, 0x13, 0x22, 0x33, 0x94, 0x88, 0x22, 0x60, 0x4f, 0xc1, 0xb7, 0xc3,
- 0x98, 0x07, 0x28, 0x9e, 0x2a, 0x28, 0x39, 0x46, 0x17, 0x31, 0x47, 0xd8, 0xde, 0x05, 0x95, 0x21,
- 0xcb, 0x13, 0x73, 0x0b, 0x80, 0xb1, 0x7b, 0x3c, 0x1e, 0xba, 0x93, 0xa3, 0xa1, 0xf7, 0xf9, 0x8d,
- 0xf5, 0xeb, 0xea, 0xfa, 0xfe, 0xf6, 0x7d, 0x7b, 0x49, 0xd3, 0x2f, 0x3c, 0x42, 0xd6, 0x8f, 0xd4,
- 0x04, 0xed, 0x25, 0xfb, 0xb3, 0x3f, 0xd0, 0xb1, 0x6f, 0x0c, 0xf0, 0x75, 0x44, 0xcf, 0x09, 0x2e,
- 0x0f, 0x36, 0x7f, 0x83, 0x8a, 0xa4, 0x0b, 0xd2, 0x30, 0x5a, 0x46, 0xa7, 0xd6, 0x6b, 0xc2, 0x02,
- 0x10, 0xae, 0x00, 0xe1, 0x81, 0x16, 0x70, 0x55, 0xcc, 0xec, 0x81, 0x2a, 0x0f, 0x43, 0x49, 0xb2,
- 0xc6, 0x5b, 0xd5, 0x60, 0x3d, 0x69, 0x98, 0xac, 0x8c, 0x5c, 0x9d, 0xb4, 0x77, 0x34, 0xfb, 0xff,
- 0x35, 0xf6, 0x9f, 0x8a, 0xbd, 0x55, 0x66, 0x0f, 0x97, 0x8c, 0x65, 0xf4, 0x3b, 0x03, 0xd4, 0xbd,
- 0x98, 0x62, 0xca, 0xa2, 0x57, 0x87, 0x37, 0x1d, 0x50, 0x15, 0x24, 0xa5, 0x1c, 0x37, 0xde, 0x6d,
- 0x5a, 0xa2, 0x83, 0xf6, 0x9e, 0xf6, 0xdd, 0x5e, 0xf3, 0xed, 0x28, 0x5f, 0xbb, 0xec, 0x2b, 0x0b,
- 0xb1, 0xb2, 0xf1, 0xa5, 0x01, 0x3e, 0x79, 0x44, 0x4a, 0xca, 0xd9, 0xa3, 0xeb, 0x3f, 0xf0, 0x31,
- 0x42, 0xc2, 0x7f, 0x9e, 0xef, 0x87, 0x08, 0x09, 0x8f, 0x2e, 0xc8, 0x0b, 0x59, 0x8a, 0x95, 0x25,
- 0x96, 0xc1, 0x3e, 0xd8, 0x7c, 0x0e, 0x83, 0x2f, 0x9e, 0x3e, 0xa6, 0xe2, 0x81, 0x46, 0x4c, 0x9e,
- 0xd4, 0x56, 0x75, 0x7f, 0xe6, 0x04, 0x55, 0x85, 0xf8, 0xf7, 0x21, 0x00, 0x00, 0xff, 0xff, 0x00,
- 0x17, 0x94, 0x23, 0x75, 0x03, 0x00, 0x00,
+ // 404 bytes of a gzipped FileDescriptorProto
+ 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xbc, 0x93, 0x4d, 0x4e, 0xdb, 0x40,
+ 0x14, 0xc7, 0xeb, 0x36, 0x4d, 0xab, 0xc9, 0xa6, 0x75, 0x1b, 0x29, 0xf1, 0xa2, 0x49, 0x2c, 0xf5,
+ 0x63, 0xd3, 0x89, 0x9c, 0x7e, 0x49, 0x91, 0xda, 0x4a, 0x81, 0x04, 0xb1, 0x22, 0xb2, 0x23, 0x90,
+ 0xd8, 0x58, 0x63, 0x66, 0x6c, 0x46, 0xb2, 0x67, 0x46, 0x1e, 0x3b, 0x81, 0x1c, 0x82, 0x43, 0x70,
+ 0x06, 0x24, 0xf6, 0x1c, 0x88, 0x03, 0xb0, 0x42, 0x1e, 0x4f, 0x50, 0x30, 0x8b, 0xc0, 0x86, 0xe5,
+ 0xcc, 0xfb, 0xbf, 0xf7, 0x7e, 0xbf, 0xc5, 0x03, 0x6d, 0x99, 0x21, 0x86, 0x51, 0x8a, 0xfd, 0x05,
+ 0x65, 0x98, 0x2f, 0xfc, 0x90, 0x49, 0x28, 0x52, 0x9e, 0x71, 0xb3, 0xc7, 0xd3, 0x08, 0x22, 0x81,
+ 0x8e, 0x8e, 0x09, 0x0c, 0x08, 0x4a, 0x60, 0xc2, 0x31, 0x89, 0xa1, 0xa0, 0x82, 0xc4, 0x94, 0x11,
+ 0x38, 0x77, 0xac, 0x66, 0xf1, 0xef, 0xa7, 0x39, 0x63, 0x24, 0xf5, 0x91, 0xa0, 0x65, 0xa7, 0xf5,
+ 0x29, 0xe2, 0x3c, 0x8a, 0x49, 0x5f, 0xbd, 0x82, 0x3c, 0xec, 0xe3, 0x3c, 0x45, 0x19, 0xe5, 0x4c,
+ 0xd7, 0x3b, 0xd5, 0x7a, 0x46, 0x13, 0x22, 0x33, 0x94, 0x88, 0x32, 0x60, 0xef, 0x83, 0x8f, 0x3b,
+ 0x31, 0x0f, 0x50, 0x7c, 0xa0, 0xa0, 0xe4, 0x14, 0x9d, 0xc6, 0x1c, 0x61, 0xfb, 0x1f, 0xa8, 0x8d,
+ 0x59, 0x9e, 0x98, 0xbf, 0x01, 0x98, 0xba, 0x7b, 0xd3, 0xb1, 0x3b, 0xdb, 0x1d, 0x7b, 0xef, 0x5e,
+ 0x58, 0x5f, 0xce, 0x2f, 0x6e, 0xae, 0x5e, 0x77, 0x0b, 0x9a, 0xe1, 0x9d, 0xc7, 0x30, 0x52, 0x23,
+ 0xb4, 0x98, 0x1c, 0xce, 0x1d, 0xfb, 0xd2, 0x00, 0x1f, 0x26, 0xf4, 0x84, 0xe0, 0xfb, 0x73, 0xcd,
+ 0xef, 0xa0, 0x26, 0xe9, 0x92, 0xb4, 0x8c, 0xae, 0xf1, 0xad, 0x31, 0x68, 0xc3, 0x92, 0x0f, 0xae,
+ 0xf8, 0xe0, 0xb6, 0xe6, 0x77, 0x55, 0xcc, 0x1c, 0x80, 0x3a, 0x0f, 0x43, 0x49, 0xb2, 0xd6, 0x4b,
+ 0xd5, 0x60, 0x3d, 0x68, 0x98, 0xad, 0x84, 0x5c, 0x9d, 0xb4, 0xff, 0x6a, 0xf4, 0x5f, 0x15, 0xf4,
+ 0xcf, 0x0a, 0xbd, 0x53, 0x41, 0x0f, 0x0b, 0xc8, 0x75, 0xf2, 0x6b, 0x03, 0x34, 0xbd, 0x98, 0x62,
+ 0xca, 0xa2, 0x67, 0x67, 0x37, 0x1d, 0x50, 0x17, 0x24, 0xa5, 0x1c, 0xb7, 0x5e, 0x6d, 0x5a, 0xa2,
+ 0x83, 0xf6, 0x7f, 0xad, 0xfb, 0xa7, 0xa2, 0xfb, 0x55, 0xe9, 0xf6, 0x2a, 0xba, 0xb2, 0x34, 0x5b,
+ 0x17, 0x3e, 0x2b, 0x84, 0x89, 0x94, 0x94, 0xb3, 0x8a, 0xf0, 0x4f, 0xf0, 0x36, 0x42, 0xc2, 0x7f,
+ 0x9c, 0xf4, 0x9b, 0x08, 0x09, 0x8f, 0x2e, 0xc9, 0x53, 0x81, 0xca, 0xcd, 0x6b, 0x40, 0xa3, 0x2d,
+ 0xb0, 0xf9, 0x20, 0x46, 0xef, 0x3d, 0x7d, 0x4e, 0x25, 0xf3, 0x84, 0xc9, 0xc3, 0xc6, 0xaa, 0xee,
+ 0xcf, 0x9d, 0xa0, 0xae, 0x08, 0x7f, 0xdc, 0x06, 0x00, 0x00, 0xff, 0xff, 0xb9, 0x67, 0xf3, 0xc1,
+ 0x77, 0x03, 0x00, 0x00,
}
diff --git a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java
index a3e959b..f6e6553 100644
--- a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java
+++ b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java
@@ -360,7 +360,7 @@ public class ProcessBundleHandlerTest {
.setWindowCoderId("window-strategy-coder")
.setWindowFn(
RunnerApi.FunctionSpec.newBuilder()
- .setUrn("beam:windowfn:global_windows:v0.1"))
+ .setUrn("beam:window_fn:global_windows:v1"))
.setOutputTime(RunnerApi.OutputTime.Enum.END_OF_WINDOW)
.setAccumulationMode(RunnerApi.AccumulationMode.Enum.ACCUMULATING)
.setTrigger(
diff --git a/sdks/python/apache_beam/portability/common_urns.py b/sdks/python/apache_beam/portability/common_urns.py
index 6f2943e..e21f04b 100644
--- a/sdks/python/apache_beam/portability/common_urns.py
+++ b/sdks/python/apache_beam/portability/common_urns.py
@@ -31,7 +31,7 @@ from apache_beam.portability.api.metrics_pb2_urns import MonitoringInfoSpecs
from apache_beam.portability.api.metrics_pb2_urns import MonitoringInfoTypeUrns
from apache_beam.portability.api.standard_window_fns_pb2_urns import FixedWindowsPayload
from apache_beam.portability.api.standard_window_fns_pb2_urns import GlobalWindowsPayload
-from apache_beam.portability.api.standard_window_fns_pb2_urns import SessionsPayload
+from apache_beam.portability.api.standard_window_fns_pb2_urns import SessionWindowsPayload
from apache_beam.portability.api.standard_window_fns_pb2_urns import SlidingWindowsPayload
primitives = StandardPTransforms.Primitives
@@ -49,7 +49,7 @@ environments = StandardEnvironments.Environments
global_windows = GlobalWindowsPayload.Enum.PROPERTIES
fixed_windows = FixedWindowsPayload.Enum.PROPERTIES
sliding_windows = SlidingWindowsPayload.Enum.PROPERTIES
-session_windows = SessionsPayload.Enum.PROPERTIES
+session_windows = SessionWindowsPayload.Enum.PROPERTIES
monitoring_info_specs = MonitoringInfoSpecs.Enum
monitoring_info_types = MonitoringInfoTypeUrns.Enum
diff --git a/sdks/python/apache_beam/portability/python_urns.py b/sdks/python/apache_beam/portability/python_urns.py
index 358c9b3..6661afe 100644
--- a/sdks/python/apache_beam/portability/python_urns.py
+++ b/sdks/python/apache_beam/portability/python_urns.py
@@ -26,7 +26,7 @@ PICKLED_DOFN_INFO = "beam:dofn:pickled_python_info:v1"
PICKLED_SOURCE = "beam:source:pickled_python:v1"
PICKLED_TRANSFORM = "beam:transform:pickled_python:v1"
PICKLED_WINDOW_MAPPING_FN = "beam:window_mapping_fn:pickled_python:v1"
-PICKLED_WINDOWFN = "beam:windowfn:pickled_python:v1"
+PICKLED_WINDOWFN = "beam:window_fn:pickled_python:v1"
PICKLED_VIEWFN = "beam:view_fn:pickled_python_data:v1"
IMPULSE_READ_TRANSFORM = "beam:transform:read_from_impulse_python:v1"
diff --git a/sdks/python/apache_beam/transforms/window.py b/sdks/python/apache_beam/transforms/window.py
index 4d900a5..4889d34 100644
--- a/sdks/python/apache_beam/transforms/window.py
+++ b/sdks/python/apache_beam/transforms/window.py
@@ -585,13 +585,14 @@ class Sessions(WindowFn):
def to_runner_api_parameter(self, context):
return (
common_urns.session_windows.urn,
- standard_window_fns_pb2.SessionsPayload(
+ standard_window_fns_pb2.SessionWindowsPayload(
gap_size=proto_utils.from_micros(
duration_pb2.Duration, self.gap_size.micros)))
@staticmethod
@urns.RunnerApiFn.register_urn(
- common_urns.session_windows.urn, standard_window_fns_pb2.SessionsPayload)
+ common_urns.session_windows.urn,
+ standard_window_fns_pb2.SessionWindowsPayload)
def from_runner_api_parameter(fn_parameter, unused_context):
return Sessions(
gap_size=Duration(micros=fn_parameter.gap_size.ToMicroseconds()))