Skip to content

[flink] improve lifecycle handling of GroupAlsoByWindowWrapper #94

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
asfgit merged 2 commits into from
Mar 31, 2016
Merged

[flink] improve lifecycle handling of GroupAlsoByWindowWrapper #94

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
033b924
[flink] improve lifecycle handling of GroupAlsoByWindowWrapper
mxm Mar 28, 2016
63a7c3d
[flink] improve readability of processElement function
mxm Mar 30, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 14 additions & 8 deletions ...ache/beam/runners/flink/translation/wrappers/streaming/FlinkGroupAlsoByWindowWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,7 @@ private FlinkGroupAlsoByWindowWrapper(PipelineOptions options,
public void open() throws Exception {
super.open();
this.context = new ProcessContext(operator, new TimestampedCollector<>(output), this.timerInternals);
operator.startBundle(context);
}

/**
Expand Down Expand Up @@ -252,19 +253,23 @@ private <W extends BoundedWindow> DoFn<KeyedWorkItem<K, VIN>, KV<K, VOUT>> creat

private void processKeyedWorkItem(KeyedWorkItem<K, VIN> workItem) throws Exception {
context.setElement(workItem, getStateInternalsForKey(workItem.key()));

// TODO: Ideally startBundle/finishBundle would be called when the operator is first used / about to be discarded.
operator.startBundle(context);
operator.processElement(context);
operator.finishBundle(context);
}

@Override
public void processElement(StreamRecord<WindowedValue<KV<K, VIN>>> element) throws Exception {
ArrayList<WindowedValue<VIN>> elements = new ArrayList<>();
elements.add(WindowedValue.of(element.getValue().getValue().getValue(), element.getValue().getTimestamp(),
element.getValue().getWindows(), element.getValue().getPane()));
processKeyedWorkItem(KeyedWorkItems.elementsWorkItem(element.getValue().getValue().getKey(), elements));
final WindowedValue<KV<K, VIN>> windowedValue = element.getValue();
final KV<K, VIN> kv = windowedValue.getValue();

final WindowedValue<VIN> updatedWindowedValue = WindowedValue.of(kv.getValue(),
windowedValue.getTimestamp(),
windowedValue.getWindows(),
windowedValue.getPane());

processKeyedWorkItem(
KeyedWorkItems.elementsWorkItem(
kv.getKey(),
Collections.singletonList(updatedWindowedValue)));
}

@Override
Expand Down Expand Up @@ -309,6 +314,7 @@ public void processWatermark(Watermark mark) throws Exception {

@Override
public void close() throws Exception {
operator.finishBundle(context);
super.close();
}

Expand Down