Nwang/add custom operator refactor #3049
Conversation
There was a problem hiding this comment.
I provided some feedback based on my limited knowledge of the streamlet part.
@srkukarni @jerrypeng Could you guys also take a look if you have some time?
| @@ -22,6 +22,8 @@ | |||
| import java.util.List; | |||
|
|
|||
| import org.apache.heron.classification.InterfaceStability; | |||
| import org.apache.heron.streamlet.impl.operators.IStreamletBasicOperator; | |||
| import org.apache.heron.streamlet.impl.operators.IStreamletOperator; | |||
There was a problem hiding this comment.
interface class should be placed in directory streamlet instead of streamlet/impl.
Only concrete implementations should be put into the impl directory.
There was a problem hiding this comment.
Make sense. Will do.
| * @param <T> The return type of the transform | ||
| * @return Streamlet containing the output of the operation | ||
| */ | ||
| <T> Streamlet<T> perform(IStreamletOperator<R, T> operator); |
There was a problem hiding this comment.
perform is not very intuitive for the meaning of this method's actual action. Let's come up with another name. How about applyOperator?
There was a problem hiding this comment.
Yao and I thought about a few options and perform is the best we got.
apply() indeed is better, however it is a special function in Java/Scala so it cant be used. applyOperator() is valid, but maybe the "Operator" part sounds a bit redundant?
I am totally open for a better name. Let's see other thoughts.
| * @param op The user defined CustomeBasicOperator | ||
| */ | ||
| public CustomBasicStreamlet(StreamletImpl<R> parent, | ||
| IStreamletBasicOperator<R, T> op) { |
There was a problem hiding this comment.
for custom operator, grouping strategy should also be provided
There was a problem hiding this comment.
Yeah. Grouping is definitely needed.
It will do in the next PR. After the grouping refactor PR is merged. #3040
| @@ -22,6 +22,8 @@ | |||
| import java.util.List; | |||
|
|
|||
| import org.apache.heron.classification.InterfaceStability; | |||
| import org.apache.heron.streamlet.impl.operators.IStreamletBasicOperator; | |||
There was a problem hiding this comment.
Given all the existing StreamletOperator extends BaseRichBolt, should we also enforce custom operator to be BaseRichBolt?
Thus we can reduce this redundant IStreamletBasicOperator.
There was a problem hiding this comment.
Yesh. I think we should enforce all StreamletOperator to be IRichBolt. CustomOperator will be the base class for all user defined operators (non-windowed onces) hence they will be IRichBolt.
IStreamletBasicOperator is not for CustomOperator. It is for reusing existing user Bolts (the ones extend from IBasicBolt instead of IRichBolt) only. That is why we have only IStreamletBasicOperator but no CustomBasicOperator.
| import org.apache.heron.api.bolt.IRichBolt; | ||
|
|
||
| /** | ||
| * The interface for custom operators: including new user defined operators as well as |
There was a problem hiding this comment.
The comment here is inconsistent with the actual code.
This interface is actually for all the operators not only for the custom operators.
Need to update this comment or the code to keep them consistent.
There was a problem hiding this comment.
Good catch. Will do.
| @@ -22,6 +22,8 @@ | |||
| import java.util.List; | |||
|
|
|||
| import org.apache.heron.classification.InterfaceStability; | |||
There was a problem hiding this comment.
I think we may also need to add CustomSource to allow users to provide specific spout as the input?
There was a problem hiding this comment.
Yeah. It is in another PR already. #3032 (may need refactor though)
| /** | ||
| * CustomOperator is the base class for all user defined operators. | ||
| * Usage: | ||
| * 1. Create user defined operator |
There was a problem hiding this comment.
This part needs further discussion. IMO, users should only need to provide the Bolt to the streamlet API instead of having direct access to streamlet's operator implementation and being required to implement an additional process method.
Once the desired bolt is provided, streamlet should take care of the remaining stuff.
There was a problem hiding this comment.
It is not an addition method users need to implement.
This class is for all "new" user defined streamlet operators and there is no Bolt concept in it.
Existing Bolts (or write operators in low level way) is supported but it another different use case.
class UserBoltOperator
extends UserBolt
implements ICustomOperator<InputType, OutputType> {
}
Then the bolt is converted to a streamlet operator and can be used in Streamlet API.
source
....
.perform(new UserBoltOperator())
....
This is the implementation of CustomOperator based on this design doc:
https://docs.google.com/document/d/1XzF0IlfuaaW8Gx3cPx1xLtP-kgCFK0TRNS5aAzuMuMg/edit#
Todos in future PRs:
Existing logic doesn't change.