You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by sandeshh <gi...@git.apache.org> on 2016/01/06 02:17:17 UTC
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
GitHub user sandeshh opened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/159
*Review only* N-DelayOperator
@davidyan74 please review.
This is the basic implementation of N-DelayOperator using WindowDataManager for iteration.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/sandeshh/incubator-apex-malhar apex247
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/159.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #159
----
commit 8f66503e39ec25a83ca57d14cdfa89b4ce7ce090
Author: sandeshh <sa...@gmail.com>
Date: 2016-01-06T01:15:32Z
N-DelayOperator
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r49011262
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public nDelayOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public nDelayOperator(int delay)
+ {
+ if ( delay < 1 ) {
+ throw new IllegalArgumentException("Invalid Delay specified.");
+ }
+ this.delay = delay;
+ init();
+ }
+
+ /*
+* Get Window Data Manager instance
+*/
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ /*
+ * Set Window Data Manager instance
+ */
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private void init()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow()
+ {
+ replay(currentWindowId - delay);
+ }
+
+ private void replay( long windowId )
+ {
+ if ( windowId < 0 ) {
+ return;
+ }
+
+ ArrayList<T> recoveredData;
+ try {
+ recoveredData = (ArrayList<T>)this.windowDataManager.load(operatorContextId, windowId);
+ if (recoveredData == null) {
+ return;
+ }
+ for ( T tuple : recoveredData) {
+ output.emit(tuple);
+ }
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ currentWindowId = windowId;
+
+ if ( delay > 1 ) {
+ replay(windowId - delay + 1);
+ }
+ }
+
+ @Override
+ public void endWindow()
+ {
+ try {
+ this.windowDataManager.save(windowData, operatorContextId, currentWindowId);
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+
+ windowData.clear();
+ }
+
+ protected void processTuple(T tuple)
+ {
+ windowData.add(tuple);
+
+ if ( delay == 1 ) {
+ output.emit(tuple);
+ }
--- End diff --
What happens if delay > 1?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by sandeshh <gi...@git.apache.org>.
Github user sandeshh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r49014450
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public nDelayOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public nDelayOperator(int delay)
+ {
+ if ( delay < 1 ) {
+ throw new IllegalArgumentException("Invalid Delay specified.");
+ }
+ this.delay = delay;
+ init();
+ }
+
+ /*
+* Get Window Data Manager instance
+*/
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ /*
+ * Set Window Data Manager instance
+ */
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private void init()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow()
+ {
+ replay(currentWindowId - delay);
+ }
+
+ private void replay( long windowId )
+ {
+ if ( windowId < 0 ) {
+ return;
+ }
+
+ ArrayList<T> recoveredData;
+ try {
+ recoveredData = (ArrayList<T>)this.windowDataManager.load(operatorContextId, windowId);
+ if (recoveredData == null) {
+ return;
+ }
+ for ( T tuple : recoveredData) {
+ output.emit(tuple);
+ }
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ currentWindowId = windowId;
+
+ if ( delay > 1 ) {
+ replay(windowId - delay + 1);
+ }
+ }
+
+ @Override
+ public void endWindow()
+ {
+ try {
+ this.windowDataManager.save(windowData, operatorContextId, currentWindowId);
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+
+ windowData.clear();
+ }
+
+ protected void processTuple(T tuple)
+ {
+ windowData.add(tuple);
+
+ if ( delay == 1 ) {
+ output.emit(tuple);
+ }
--- End diff --
Tuples are just stored in WindowDataManager and replayed during BeginWindow. So it is important not to change delay after the object is constructed, so it can only be set through constructor.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r50044513
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public nDelayOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public nDelayOperator(int delay)
+ {
+ if ( delay < 1 ) {
+ throw new IllegalArgumentException("Invalid Delay specified.");
+ }
+ this.delay = delay;
+ init();
+ }
+
+ /*
+* Get Window Data Manager instance
+*/
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ /*
+ * Set Window Data Manager instance
+ */
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private void init()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow()
+ {
+ replay(currentWindowId - delay);
+ }
+
+ private void replay( long windowId )
+ {
+ if ( windowId < 0 ) {
+ return;
+ }
+
+ ArrayList<T> recoveredData;
+ try {
+ recoveredData = (ArrayList<T>)this.windowDataManager.load(operatorContextId, windowId);
+ if (recoveredData == null) {
+ return;
+ }
+ for ( T tuple : recoveredData) {
+ output.emit(tuple);
+ }
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ currentWindowId = windowId;
+
+ if ( delay > 1 ) {
+ replay(windowId - delay + 1);
+ }
+ }
+
+ @Override
+ public void endWindow()
+ {
+ try {
+ this.windowDataManager.save(windowData, operatorContextId, currentWindowId);
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+
+ windowData.clear();
+ }
+
+ protected void processTuple(T tuple)
+ {
+ windowData.add(tuple);
+
+ if ( delay == 1 ) {
+ output.emit(tuple);
+ }
+ }
+
+ @Override
+ public void checkpointed(long l)
+ {
+
+ }
+
+ @Override
+ public void committed(long windowId)
+ {
+ try {
+ windowDataManager.deleteUpTo(operatorContextId, windowId - delay - 1);
--- End diff --
windowId arithmetic needs to be changed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by sandeshh <gi...@git.apache.org>.
Github user sandeshh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r49147546
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public nDelayOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public nDelayOperator(int delay)
+ {
+ if ( delay < 1 ) {
+ throw new IllegalArgumentException("Invalid Delay specified.");
+ }
+ this.delay = delay;
+ init();
+ }
+
+ /*
+* Get Window Data Manager instance
+*/
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ /*
+ * Set Window Data Manager instance
+ */
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private void init()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow()
+ {
+ replay(currentWindowId - delay);
--- End diff --
Making the necessary changes. It also means cleanup logic needs to be handled carefully.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r50052415
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/DelayByNOperator.java ---
@@ -0,0 +1,225 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+import com.datatorrent.stram.engine.WindowGenerator;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class DelayByNOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+ private long previouslyCommittedWindow;
+ private boolean previouslyCommittedWindowSet = false;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public DelayByNOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public DelayByNOperator(int delay)
+ {
+ if ( delay < 1 ) {
--- End diff --
When the logic is done, please correct the code style so it can be merged after https://github.com/apache/incubator-apex-core/pull/185 is merged.
- no space after open parenthesis and before close parenthesis.
- classes (including unit test classes) must be capitalized.
- (someBoolean == false) should be changed to (!someBoolean)
- all open braces that denotes a definition must be in its own line
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r49010907
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public nDelayOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public nDelayOperator(int delay)
+ {
+ if ( delay < 1 ) {
+ throw new IllegalArgumentException("Invalid Delay specified.");
+ }
+ this.delay = delay;
+ init();
+ }
+
+ /*
+* Get Window Data Manager instance
+*/
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ /*
+ * Set Window Data Manager instance
+ */
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private void init()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow()
+ {
+ replay(currentWindowId - delay);
--- End diff --
We cannot simply subtract from windowId since there is a baseSecond rollover mechanism. Refer to WindowGenerator.getAheadWindowId()
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r49010996
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public nDelayOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public nDelayOperator(int delay)
+ {
+ if ( delay < 1 ) {
+ throw new IllegalArgumentException("Invalid Delay specified.");
+ }
+ this.delay = delay;
+ init();
+ }
+
+ /*
+* Get Window Data Manager instance
+*/
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ /*
+ * Set Window Data Manager instance
+ */
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private void init()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow()
+ {
+ replay(currentWindowId - delay);
+ }
+
+ private void replay( long windowId )
+ {
+ if ( windowId < 0 ) {
+ return;
+ }
+
+ ArrayList<T> recoveredData;
+ try {
+ recoveredData = (ArrayList<T>)this.windowDataManager.load(operatorContextId, windowId);
+ if (recoveredData == null) {
+ return;
+ }
+ for ( T tuple : recoveredData) {
+ output.emit(tuple);
+ }
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
--- End diff --
rethrow() is deprecated.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by PramodSSImmaneni <gi...@git.apache.org>.
Github user PramodSSImmaneni commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r48918818
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public nDelayOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public nDelayOperator(int delay)
+ {
+ if ( delay < 1 ) {
+ throw new IllegalArgumentException("Invalid Delay specified.");
+ }
+ this.delay = delay;
+ init();
+ }
+
+ /*
+* Get Window Data Manager instance
+*/
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ /*
+ * Set Window Data Manager instance
+ */
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private void init()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow()
+ {
+ replay(currentWindowId - delay);
+ }
+
+ private void replay( long windowId )
+ {
+ if ( windowId < 0 ) {
+ return;
+ }
+
+ ArrayList<T> recoveredData;
+ try {
+ recoveredData = (ArrayList<T>)this.windowDataManager.load(operatorContextId, windowId);
+ if (recoveredData == null) {
+ return;
+ }
+ for ( T tuple : recoveredData) {
+ output.emit(tuple);
+ }
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ currentWindowId = windowId;
+
+ if ( delay > 1 ) {
+ replay(windowId - delay + 1);
--- End diff --
I don't think operator should be doing this. We should enhance the platform to do the n-delay by incrementing the window id by n for data transmitted in a window. This should have to be managed within the operator where operator is doing (n-1) and platform is doing +1. My suggestion is to implement n delay in platform and take this operator after. Also when that happens the operator needs to only save data from (checkpoint-n, checkpoint] window range.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by chandnisingh <gi...@git.apache.org>.
Github user chandnisingh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r48918635
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
--- End diff --
maybe DelayByNOperator?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by sandeshh <gi...@git.apache.org>.
Github user sandeshh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r48920790
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
--- End diff --
will use that name.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by chandnisingh <gi...@git.apache.org>.
Github user chandnisingh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r48918526
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
--- End diff --
Sorry for nitpicking but the name is not according to Java Convention of naming classes
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r50052538
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/DelayByNOperator.java ---
@@ -0,0 +1,225 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+import com.datatorrent.stram.engine.WindowGenerator;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class DelayByNOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+ private long previouslyCommittedWindow;
+ private boolean previouslyCommittedWindowSet = false;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public DelayByNOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public DelayByNOperator(int delay)
+ {
+ if ( delay < 1 ) {
+ throw new IllegalArgumentException("Invalid Delay specified.");
+ }
+ this.delay = delay;
+ init();
+ }
+
+ /*
+* Get Window Data Manager instance
+*/
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ /*
+ * Set Window Data Manager instance
+ */
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private void init()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow()
+ {
+ replay(currentWindowId, delay, true);
+ }
+
+ private void replay( long windowId, int delay, boolean firstWindow )
+ {
+ if ( WindowGenerator.MIN_WINDOW_ID + delay > windowId ) {
+ windowId = WindowGenerator.MAX_WINDOW_ID - ( delay - ( windowId - WindowGenerator.MIN_WINDOW_ID ) ) + 1;
+ } else {
+ windowId = windowId - delay;
+ }
+
+ if ( !firstWindow) {
+ ++windowId;
+ }
+
+ ArrayList<T> recoveredData;
+ try {
+ recoveredData = (ArrayList<T>)this.windowDataManager.load(operatorContextId, windowId);
+ if (recoveredData == null) {
+ return;
+ }
+ for ( T tuple : recoveredData) {
+ output.emit(tuple);
+ }
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
--- End diff --
rethrow is deprecated
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r49014853
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public nDelayOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public nDelayOperator(int delay)
+ {
+ if ( delay < 1 ) {
+ throw new IllegalArgumentException("Invalid Delay specified.");
+ }
+ this.delay = delay;
+ init();
+ }
+
+ /*
+* Get Window Data Manager instance
+*/
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ /*
+ * Set Window Data Manager instance
+ */
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private void init()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow()
+ {
+ replay(currentWindowId - delay);
+ }
+
+ private void replay( long windowId )
+ {
+ if ( windowId < 0 ) {
+ return;
+ }
+
+ ArrayList<T> recoveredData;
+ try {
+ recoveredData = (ArrayList<T>)this.windowDataManager.load(operatorContextId, windowId);
+ if (recoveredData == null) {
+ return;
+ }
+ for ( T tuple : recoveredData) {
+ output.emit(tuple);
+ }
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ currentWindowId = windowId;
+
+ if ( delay > 1 ) {
+ replay(windowId - delay + 1);
+ }
+ }
+
+ @Override
+ public void endWindow()
+ {
+ try {
+ this.windowDataManager.save(windowData, operatorContextId, currentWindowId);
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+
+ windowData.clear();
+ }
+
+ protected void processTuple(T tuple)
+ {
+ windowData.add(tuple);
+
+ if ( delay == 1 ) {
+ output.emit(tuple);
+ }
--- End diff --
I see.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r50044558
--- Diff: library/src/test/java/com/datatorrent/lib/iteration/nDelayOperatorTest.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import com.datatorrent.api.*;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.helper.OperatorContextTestHelper;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.testbench.RandomEventGenerator;
+import com.datatorrent.lib.util.WindowDataManager;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+
+public class nDelayOperatorTest {
+
+ public static OperatorContextTestHelper.TestIdOperatorContext context;
+
+ @Rule
+ public TestMeta testMeta = new TestMeta();
+
+ public static class TestMeta extends TestWatcher
+ {
+
+ @Override
+ protected void starting(org.junit.runner.Description description)
+ {
+ String methodName = description.getMethodName();
+ String className = description.getClassName();
+
+ Attribute.AttributeMap.DefaultAttributeMap attributes = new Attribute.AttributeMap.DefaultAttributeMap();
+ attributes.put(Context.DAGContext.APPLICATION_PATH,
+ "target/" + className + "/" + methodName + "/" );
+
+ attributes.put(DAG.APPLICATION_ID, "appId");
+
+ context = new OperatorContextTestHelper.TestIdOperatorContext(0, attributes) ;
+ }
+
+ @Override
+ protected void finished(Description description)
+ {
+ FileUtils.deleteQuietly(new File("target/" + description.getClassName() + "/" + description.getMethodName()));
+ }
+ }
+
+ @Test
+ public void testIntegrationWithWindowDataManager() throws Exception {
+
+ nDelayOperator<Integer> nDelayOperator = new nDelayOperator<>(2) ;
+ nDelayOperator.setWindowDataManager(new WindowDataManager.FSWindowDataManager());
+
+ nDelayOperator.setup(context);
+
+ CollectorTestSink testSink = new CollectorTestSink();
+ nDelayOperator.output.setSink(testSink);
+
+ nDelayOperator.beginWindow(1);
+ nDelayOperator.input.process(1);
+ nDelayOperator.input.process(2);
+ nDelayOperator.endWindow();
+
+ nDelayOperator.beginWindow(2);
+ nDelayOperator.input.process(3);
+ nDelayOperator.input.process(4);
+ nDelayOperator.endWindow();
+
+ nDelayOperator.beginWindow(3);
+ nDelayOperator.endWindow();
+
+ nDelayOperator.setWindowDataManager(new WindowDataManager.FSWindowDataManager());
+
+ nDelayOperator.setup(context);
+
+ testSink = new CollectorTestSink();
+ nDelayOperator.output.setSink(testSink);
+
+ nDelayOperator.firstWindow();
+
+ Assert.assertEquals("required tuples - replayed", testSink.collectedTuples.get(0), 1);
+ Assert.assertEquals("required tuples - replayed", testSink.collectedTuples.get(1), 2);
+
+ nDelayOperator.beginWindow(4);
+ nDelayOperator.endWindow();
+
+ testSink.clear();
+
+ nDelayOperator.firstWindow();
+
+ Assert.assertEquals("required tuples - replayed", testSink.collectedTuples.get(0), 3);
+ Assert.assertEquals("required tuples - replayed", testSink.collectedTuples.get(1), 4);
+ }
+
+ @Test
+ public void testWindowDelay() throws Exception
+ {
+ for ( int i = 1 ; i < 5 ; ++i ) {
+ LocalMode lma = LocalMode.newInstance();
+ Application.delay = i;
+ Configuration conf = new Configuration(false);
+ lma.prepareDAG(new Application(), conf);
+ LocalMode.Controller lc = lma.getController();
+ lc.run(10000);
+ }
+ }
+
+ @ApplicationAnnotation(name="IterationDemo")
+ public static class Application implements StreamingApplication
+ {
+ private final static Logger LOG = LoggerFactory.getLogger(Application.class);
+ public static int delay = 1;
+
+ public static class OutputOperator extends BaseOperator
+ {
+ public long windowId ;
+
+ public transient DefaultInputPort<Long> input = new DefaultInputPort<Long>()
+ {
+ @Override
+ public void process(Long tuple)
+ {
+
+ Assert.assertEquals("number emitted tuples", windowId - delay, tuple.longValue());
+ }
+ };
+
+ public void beginWindow(long windowId)
+ {
+ this.windowId = windowId;
+ }
+
+ public transient DefaultInputPort<Integer> inputDummy = new DefaultInputPort<Integer>()
+ {
+ @Override
+ public void process(Integer tuple)
+ {
+ output.emit(windowId);
+ }
+ };
+ public transient DefaultOutputPort<Long> output = new DefaultOutputPort<>() ;
+
+ }
+
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ RandomEventGenerator rand = dag.addOperator("rand", new RandomEventGenerator());
+ rand.setTuplesBlast(1);
+ rand.setTuplesBlastIntervalMillis(1000);
+ nDelayOperator<Long> nDelayOperator = dag.addOperator("delay", new nDelayOperator<Long>(delay));
+ OutputOperator outputOperator = dag.addOperator("output", new OutputOperator());
+
+ nDelayOperator.setWindowDataManager(new WindowDataManager.NoopWindowDataManager());
+
+ dag.addStream("rand_inputToDelay", rand.integer_data, outputOperator.inputDummy);
+ dag.addStream("inputToDelay_toDelay", outputOperator.output, nDelayOperator.input);
+ dag.addStream("DelayOperator_output", nDelayOperator.output, outputOperator.input);
+ }
+ }
+}
--- End diff --
Can you add a couple of test cases that test the window id baseSecond rollover?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/159#discussion_r49011370
--- Diff: library/src/main/java/com/datatorrent/lib/iteration/nDelayOperator.java ---
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.iteration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import javax.validation.constraints.Min;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.util.WindowDataManager;
+import com.datatorrent.netlet.util.DTThrowable;
+
+/**
+ * This operator is used in iteration and can do n window delay.
+ * Required delay needs to be set in the constructor, default delay is 1.
+ *
+ * @displayName nDelayOperator
+ * @category iteration
+
+ */
+public class nDelayOperator<T> implements Operator.DelayOperator, Operator.CheckpointListener
+{
+ private WindowDataManager windowDataManager = new WindowDataManager.FSWindowDataManager();
+ @Min(1)
+ private int delay = 1;
+ private long currentWindowId;
+ private transient int operatorContextId;
+ private transient ArrayList<T> windowData;
+ private transient Context.OperatorContext context;
+
+ public transient DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t)
+ {
+ processTuple(t);
+ }
+ };
+
+ public transient DefaultOutputPort<T> output = new DefaultOutputPort();
+
+ public nDelayOperator()
+ {
+ init();
+ }
+
+ /*
+ * @param delay set the delay in number of windows for the tuples.
+ */
+ public nDelayOperator(int delay)
+ {
+ if ( delay < 1 ) {
+ throw new IllegalArgumentException("Invalid Delay specified.");
+ }
+ this.delay = delay;
+ init();
+ }
+
+ /*
+* Get Window Data Manager instance
+*/
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ /*
+ * Set Window Data Manager instance
+ */
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private void init()
+ {
+ windowData = new ArrayList<>();
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ this.operatorContextId = context.getId();
+ this.windowDataManager.setup(context);
+ this.context = context;
+ }
+
+ @Override
+ public void teardown()
+ {
+ this.windowDataManager.teardown();
+ }
+
+ @Override
+ public void firstWindow()
+ {
+ replay(currentWindowId - delay);
+ }
+
+ private void replay( long windowId )
+ {
+ if ( windowId < 0 ) {
+ return;
+ }
+
+ ArrayList<T> recoveredData;
+ try {
+ recoveredData = (ArrayList<T>)this.windowDataManager.load(operatorContextId, windowId);
+ if (recoveredData == null) {
+ return;
+ }
+ for ( T tuple : recoveredData) {
+ output.emit(tuple);
+ }
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ currentWindowId = windowId;
+
+ if ( delay > 1 ) {
+ replay(windowId - delay + 1);
--- End diff --
We cannot do simple arithmetic on window id because of baseSecond rollover. See my comment above.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: *Review only* N-DelayOperator
Posted by sandeshh <gi...@git.apache.org>.
Github user sandeshh closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/159
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---