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.
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
[NEMO-252] Fix CreatViewTransform to emit windowed materialized data #141
Conversation
create view create view fix final
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks @taegeonum. I've left comments.
private final MultiView<Object> multiView; | ||
private final Map<BoundedWindow, List<I>> windowListMap; | ||
|
||
// TODO #XXX: we should remove this variable by refactoring broadcast worker for side input |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please create a JIRA issue.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
done.
iterator.remove(); | ||
isEmitted = true; | ||
|
||
if (outputTimestamp > entry.getKey().maxTimestamp().getMillis()) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
outputTimeStamp = Math.max(outputTimeStamp, entry.getKey().maxTimestamp().getMillis()) ?
} | ||
|
||
final Iterator<Map.Entry<BoundedWindow, List<I>>> iterator = windowListMap.entrySet().iterator(); | ||
long outputTimestamp = Long.MAX_VALUE; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
maxOutputTimeStampOfEmittedWindows?
|
||
// TODO #XXX: we should remove this variable by refactoring broadcast worker for side input | ||
private boolean isEmitted = false; | ||
private long outputWatermark; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
lastEmittedWatermark?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I will change it to currentOutputWatermark
final Instant processingTime, | ||
final Instant synchronizedTime) { | ||
keyToValues.forEach((key, val) -> { | ||
long outputTimestamp = Long.MAX_VALUE; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
maxOutputTimeStampOfEmittedWindows?
|
||
for (final Map.Entry<K, List<WindowedValue<InputT>>> entry : keyToValues.entrySet()) { | ||
final K key = entry.getKey(); | ||
final List<WindowedValue<InputT>> val = entry.getValue(); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
val -> values
final CreateViewTransform<String, Integer> viewTransform = | ||
new CreateViewTransform(new SumViewFn()); | ||
|
||
final Instant ts1 = new Instant(1); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Share code with GroupByKeyAndWindowDoFnTransformTest?
|
||
@Override | ||
public Materialization<Materializations.MultimapView<Void, String>> getMaterialization() { | ||
return null; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
throw new UnSupportedOperation?
.../src/test/java/org/apache/nemo/compiler/frontend/beam/transform/CreateViewTransformTest.java
Show resolved
Hide resolved
.../src/test/java/org/apache/nemo/compiler/frontend/beam/transform/CreateViewTransformTest.java
Show resolved
Hide resolved
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@johnyangk Thanks for the review! I've addressed your comments
private final MultiView<Object> multiView; | ||
private final Map<BoundedWindow, List<I>> windowListMap; | ||
|
||
// TODO #XXX: we should remove this variable by refactoring broadcast worker for side input |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
done.
|
||
// TODO #XXX: we should remove this variable by refactoring broadcast worker for side input | ||
private boolean isEmitted = false; | ||
private long outputWatermark; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I will change it to currentOutputWatermark
JIRA: NEMO-252: Fix CreatViewTransform to emit windowed materialized data
Major changes:
CreateViewTransform
to collect windowed data and emit them by applying a view functionMinor changes
GroupByKeyAndWindowDoFnTransform
Tests for the changes:
CreateViewTransformTest
that tests materialized data in windows