taegeonum commented on a change in pull request #135: [NEMO-230] Emit collected
data when receiving watermark in GroupByKeyAndWindowTransform
URL: https://github.com/apache/incubator-nemo/pull/135#discussion_r229187844
##########
File path:
compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/GroupByKeyAndWindowDoFnTransform.java
##########
@@ -76,30 +77,59 @@ public GroupByKeyAndWindowDoFnTransform(final
Map<TupleTag<?>, Coder<?>> outputC
*/
@Override
protected DoFn wrapDoFn(final DoFn doFn) {
- timerInternalsFactory = new InMemoryTimerInternalsFactory();
+ this.stateAndTimerInternalsFactory = new StateAndTimerInternalsFactory();
// This function performs group by key and window operation
return
GroupAlsoByWindowViaWindowSetNewDoFn.create(
getWindowingStrategy(),
- new InMemoryStateInternalsFactory(),
- timerInternalsFactory,
+ stateAndTimerInternalsFactory.inMemoryStateInternalsFactory,
+ stateAndTimerInternalsFactory.inMemoryTimerInternalsFactory,
getSideInputReader(),
reduceFn,
getOutputManager(),
getMainOutputTag());
}
+ /**
+ * It collects data for each key.
+ * The collected data are emitted at {@link
GroupByKeyAndWindowDoFnTransform#onWatermark(Watermark)}
+ * @param element data element
+ */
@Override
public void onData(final WindowedValue<KV<K, InputT>> element) {
final KV<K, InputT> kv = element.getValue();
- keyToValues.putIfAbsent(kv.getKey(), new ArrayList());
+ keyToValues.putIfAbsent(kv.getKey(), new LinkedList<>());
keyToValues.get(kv.getKey()).add(element.withValue(kv.getValue()));
Review comment:
comments are added
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
