Windows are assigned to elements via the Window.into transform. They
influence grouping operations such as GroupByKey, Combine.perKey, and
Combine.globally. Looking at your example, you start with
PCollection<KV<A,B>>
Presumably via a Read or a Create. These KVs are in a global window,
so the elements are really triples (ignoring PaneInfo) of the form
(KV<A, B>, GlobalWindow, timestamp)
From what I gather, the next step you do is a
Window.into(FixedWindows.of(...)), yielding a PCollection<KV<A,B>>
whose elements are, implicitly
(KV<A, B>, IntervalWindow, timestamp)
Now you apply a GroupByKey to get elements of the form
(KV<A, Iterable<B>>, IntervalWindow, timestamp)
where there is one Iterable for each distinct key and window. You
apply a ParDo to get PCollection<X> which is of the form
(X, IntervalWindow, timestamp)
It looks like your next step is another
Window.into(FixedWindows.of(...)), yielding
(X, IntervalWindow, timestamp)
where the IntervalWindow here may be different if the parameters to
FixedWindows were different (e.g. the first was by minute, the second
by hours). If it's the same, this is a no-op. Now you apply
Combine.globally(CombineFn<X, R>) to get a PCollection<R> whose
elements are of the form
(R, IntervalWindow, timestamp)
where there is now one R per window (the elements in the same window
being combined, the elements across windows not).
FWIW, internally, Combine.globally is implemented as PariWithNullKey +
CombinePerKey + StripNullKey.
Does this help?
On Tue, Mar 5, 2019 at 8:09 PM Daniel Debrunner <[email protected]> wrote:
Thanks for the reply.
As for every element is always associated with a window, when a
element is produced due to a window trigger (e.g. the GroupByKey) what
window is it associated with? The window it was produced from? Maybe
the question is when is a window assigned to an element?
I'll see if I can come up with an example,
Thanks,
Dan.
On Tue, Mar 5, 2019 at 10:47 AM Kenneth Knowles <[email protected]> wrote:
Two pieces to this:
1. Every element in a PCollection is always associated with a window, and
GroupByKey (hence CombinePerKey) operates per-key-and-window (w/ window
merging).
2. If an element is not explicitly a KV, then there is no key associated with
it.
I'm afraid I don't have any guesses at the problem based on what you've shared.
Can you say more?
Kenn
On Tue, Mar 5, 2019 at 10:29 AM Daniel Debrunner <[email protected]> wrote:
The windowing section of the Beam programming model guide shows a
window defined and used in the GropyByKey transform after a ParDo.
(section 7.1.1).
However I couldn't find any documentation on how long the window
remains in scope for subsequent transforms.
I have an application with this pipeline:
PCollection<KV<A,B>> -> FixedWindow<KV<A,B>> -> GroupByKey ->
PCollection<X> -> FixedWindow<X> -> Combine<X,R>.globally ->
PCollection<R>
The idea is that the first window is aggregating by key but in the
second window I need to combine elements across all keys.
With my initial app I was seeing some runtime errors in/after the
combine where a KV<null,R> was being seen, even though at that point
there should be no key for the PCollection<R>.
In a simpler test I can apply FixedWindow<X> -> Combine<X,R>.globally
-> PCollection<R> to a PCollection without an upstream window and the
combine correctly happens once.
But then adding the keyed upstream window, the combine occurs once per
key without any final combine across the keys.
So it seems somehow the memory of the key exists even with the new
window transform,
I'm probably misunderstanding some detail of windowing, but I couldn't
find any deeper documentation than the simple examples in the
programming model guide.
Can anyone point me in the correct direction?
Thanks,
Dan.
