You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by Sandesh Hegde <sa...@datatorrent.com> on 2016/04/08 19:36:06 UTC
Proposal: Variation for the Apex Api
Hello Team,
As we were talking about High Level api in another thread, here is a small
variation proposal for our existing apis,
Take look at the current code and the proposed code
Current Style
---------------------------
KafkaSinglePortStringInputOperator kafkaInput =
dag.addOperator("kafkaInput", new KafkaSinglePortStringInputOperator());
DeserializeJSON deserializeJSON = dag.addOperator("deserialize", new
DeserializeJSON());
RedisJoin redisJoin = dag.addOperator("redisJoin", new RedisJoin());
CampaignProcessorWithApexWindow campaignProcessor =
dag.addOperator("campaignProcessor", new CampaignProcessorWithApexWindow());
// Connect the Ports in the Operators
dag.addStream("deserialize", kafkaInput.outputPort,
deserializeJSON.input) ;
dag.addStream("redisJoin", filterFields.output, redisJoin.input) ;
dag.addStream("output", redisJoin.output, campaignProcessor.input);
-------------------------------------
Proposed Change ( Just create Operators and connect the output port to
input ports and then add the input operators to DAG )
-------------------------------------------
KafkaSinglePortStringInputOperator kafkaSinglePortStringInputOperator =
new KafkaSinglePortStringInputOperator() ;
DeserializeJSON deserializeJSON = new DeserializeJSON();
RedisJoin redisJoin = new RedisJoin() ;
CampaignProcessor campaignProcessor = new CampaignProcessor() ;
kafkaInput.outputPort.connect(deserializeJSON.input);
filterFields.output.connect(redisJoin.input) ;
redisJoin.output.connect(campaignProcessor.input) ;
dag.add(kafkaInput);
-----------------------------------------
Things to note : Name can be generated internally or user can specify a
name property of the operator ( like Dag.AddOperator("name", ...) )
Let me know your thoughts.
Thanks
Re: Proposal: Variation for the Apex Api
Posted by Sandesh Hegde <sa...@datatorrent.com>.
Hi Siyuan,
These set of APIs will be v2 of the existing APIs. Internally it will be
converted into logical plan using the existing APIs. So existing APIs won't
be changed.
Thanks
Sandesh
On Tue, Apr 12, 2016 at 10:08 AM Siyuan Hua <si...@datatorrent.com> wrote:
> Hey Sandesh,
>
> I like your idea, but here is what I think.
> API Stability is VERY important! The API we have to build a dag has been
> used for a long time. If the new change doesn't help a lot, I won't
> recommend to change the API
>
> so +0 :)
>
> On Mon, Apr 11, 2016 at 12:02 PM, Sandesh Hegde <sa...@datatorrent.com>
> wrote:
>
> > Hello Chinmay,
> >
> > Idea is to follow natural order. Suppose you want to set attribute on a
> > port, you will have to write something like below
> >
> > dag.setInputPortAttribute(deserializeJSON.input,
> > Context.PortContext.PARTITION_PARALLEL, true);
> >
> > How about setting it directly on the ports?
> >
> deserializeJSON.input.setAttribute(Context.PortContext.PARTITION_PARALLEL,
> > true));
> >
> > Same thing goes for attributes of the Operators.
> >
> > These new set of APIs will support all the features of our
> > existing API, but in a more natural way.
> > It is not similar to High-Level API, it is just moonlighting idea about
> our
> > existing API.
> >
> > Thanks
> >
> >
> >
> >
> >
> >
> >
> > On Sun, Apr 10, 2016 at 11:20 PM Chinmay Kolhatkar <ch...@apache.org>
> > wrote:
> >
> > > I feel that adding all operators to dag makes it explicit for a
> developer
> > > who is looking at the code for the first time.
> > >
> > > Besides, one might want to do dag.getMeta on operators and ports to set
> > > some attributes.
> > > I'm not sure how much explicit will that be if we bypass adding
> operators
> > > to the DAG.
> > >
> > > Though, I totally agree with the intention of simplifying the APIs for
> > > which the proposal from Siyuan in another mail is a nice first step.
> > >
> > > Regards,
> > > Chinmay.
> > >
> > >
> > >
> > > On Mon, Apr 11, 2016 at 10:48 AM, Sandesh Hegde <
> sandesh@datatorrent.com
> > >
> > > wrote:
> > >
> > > > Yes, that is the idea.
> > > >
> > > > On Sun, Apr 10, 2016, 9:28 PM Sandeep Deshmukh <
> > sandeep@datatorrent.com>
> > > > wrote:
> > > >
> > > > > Are you suggesting we add only the input adapters to the dag and
> then
> > > all
> > > > > the connected ones are added by simply traversing the DAG?
> > > > >
> > > > > Regards,
> > > > > Sandeep
> > > > >
> > > > > On Fri, Apr 8, 2016 at 11:06 PM, Sandesh Hegde <
> > > sandesh@datatorrent.com>
> > > > > wrote:
> > > > >
> > > > > > Hello Team,
> > > > > >
> > > > > > As we were talking about High Level api in another thread, here
> is
> > a
> > > > > small
> > > > > > variation proposal for our existing apis,
> > > > > >
> > > > > > Take look at the current code and the proposed code
> > > > > >
> > > > > > Current Style
> > > > > > ---------------------------
> > > > > > KafkaSinglePortStringInputOperator kafkaInput =
> > > > > > dag.addOperator("kafkaInput", new
> > > > KafkaSinglePortStringInputOperator());
> > > > > > DeserializeJSON deserializeJSON =
> > dag.addOperator("deserialize",
> > > > new
> > > > > > DeserializeJSON());
> > > > > > RedisJoin redisJoin = dag.addOperator("redisJoin", new
> > > > RedisJoin());
> > > > > > CampaignProcessorWithApexWindow campaignProcessor =
> > > > > > dag.addOperator("campaignProcessor", new
> > > > > > CampaignProcessorWithApexWindow());
> > > > > >
> > > > > > // Connect the Ports in the Operators
> > > > > > dag.addStream("deserialize", kafkaInput.outputPort,
> > > > > > deserializeJSON.input) ;
> > > > > > dag.addStream("redisJoin", filterFields.output,
> > redisJoin.input)
> > > ;
> > > > > > dag.addStream("output", redisJoin.output,
> > > campaignProcessor.input);
> > > > > > -------------------------------------
> > > > > >
> > > > > >
> > > > > > Proposed Change ( Just create Operators and connect the output
> port
> > > to
> > > > > > input ports and then add the input operators to DAG )
> > > > > > -------------------------------------------
> > > > > > KafkaSinglePortStringInputOperator
> > > > > kafkaSinglePortStringInputOperator =
> > > > > > new KafkaSinglePortStringInputOperator() ;
> > > > > > DeserializeJSON deserializeJSON = new DeserializeJSON();
> > > > > > RedisJoin redisJoin = new RedisJoin() ;
> > > > > > CampaignProcessor campaignProcessor = new
> CampaignProcessor()
> > ;
> > > > > >
> > > > > > kafkaInput.outputPort.connect(deserializeJSON.input);
> > > > > > filterFields.output.connect(redisJoin.input) ;
> > > > > > redisJoin.output.connect(campaignProcessor.input) ;
> > > > > >
> > > > > > dag.add(kafkaInput);
> > > > > > -----------------------------------------
> > > > > >
> > > > > > Things to note : Name can be generated internally or user can
> > > specify a
> > > > > > name property of the operator ( like Dag.AddOperator("name",
> ...) )
> > > > > >
> > > > > > Let me know your thoughts.
> > > > > >
> > > > > > Thanks
> > > > > >
> > > > >
> > > >
> > >
> >
>
Re: Proposal: Variation for the Apex Api
Posted by Siyuan Hua <si...@datatorrent.com>.
Hey Sandesh,
I like your idea, but here is what I think.
API Stability is VERY important! The API we have to build a dag has been
used for a long time. If the new change doesn't help a lot, I won't
recommend to change the API
so +0 :)
On Mon, Apr 11, 2016 at 12:02 PM, Sandesh Hegde <sa...@datatorrent.com>
wrote:
> Hello Chinmay,
>
> Idea is to follow natural order. Suppose you want to set attribute on a
> port, you will have to write something like below
>
> dag.setInputPortAttribute(deserializeJSON.input,
> Context.PortContext.PARTITION_PARALLEL, true);
>
> How about setting it directly on the ports?
> deserializeJSON.input.setAttribute(Context.PortContext.PARTITION_PARALLEL,
> true));
>
> Same thing goes for attributes of the Operators.
>
> These new set of APIs will support all the features of our
> existing API, but in a more natural way.
> It is not similar to High-Level API, it is just moonlighting idea about our
> existing API.
>
> Thanks
>
>
>
>
>
>
>
> On Sun, Apr 10, 2016 at 11:20 PM Chinmay Kolhatkar <ch...@apache.org>
> wrote:
>
> > I feel that adding all operators to dag makes it explicit for a developer
> > who is looking at the code for the first time.
> >
> > Besides, one might want to do dag.getMeta on operators and ports to set
> > some attributes.
> > I'm not sure how much explicit will that be if we bypass adding operators
> > to the DAG.
> >
> > Though, I totally agree with the intention of simplifying the APIs for
> > which the proposal from Siyuan in another mail is a nice first step.
> >
> > Regards,
> > Chinmay.
> >
> >
> >
> > On Mon, Apr 11, 2016 at 10:48 AM, Sandesh Hegde <sandesh@datatorrent.com
> >
> > wrote:
> >
> > > Yes, that is the idea.
> > >
> > > On Sun, Apr 10, 2016, 9:28 PM Sandeep Deshmukh <
> sandeep@datatorrent.com>
> > > wrote:
> > >
> > > > Are you suggesting we add only the input adapters to the dag and then
> > all
> > > > the connected ones are added by simply traversing the DAG?
> > > >
> > > > Regards,
> > > > Sandeep
> > > >
> > > > On Fri, Apr 8, 2016 at 11:06 PM, Sandesh Hegde <
> > sandesh@datatorrent.com>
> > > > wrote:
> > > >
> > > > > Hello Team,
> > > > >
> > > > > As we were talking about High Level api in another thread, here is
> a
> > > > small
> > > > > variation proposal for our existing apis,
> > > > >
> > > > > Take look at the current code and the proposed code
> > > > >
> > > > > Current Style
> > > > > ---------------------------
> > > > > KafkaSinglePortStringInputOperator kafkaInput =
> > > > > dag.addOperator("kafkaInput", new
> > > KafkaSinglePortStringInputOperator());
> > > > > DeserializeJSON deserializeJSON =
> dag.addOperator("deserialize",
> > > new
> > > > > DeserializeJSON());
> > > > > RedisJoin redisJoin = dag.addOperator("redisJoin", new
> > > RedisJoin());
> > > > > CampaignProcessorWithApexWindow campaignProcessor =
> > > > > dag.addOperator("campaignProcessor", new
> > > > > CampaignProcessorWithApexWindow());
> > > > >
> > > > > // Connect the Ports in the Operators
> > > > > dag.addStream("deserialize", kafkaInput.outputPort,
> > > > > deserializeJSON.input) ;
> > > > > dag.addStream("redisJoin", filterFields.output,
> redisJoin.input)
> > ;
> > > > > dag.addStream("output", redisJoin.output,
> > campaignProcessor.input);
> > > > > -------------------------------------
> > > > >
> > > > >
> > > > > Proposed Change ( Just create Operators and connect the output port
> > to
> > > > > input ports and then add the input operators to DAG )
> > > > > -------------------------------------------
> > > > > KafkaSinglePortStringInputOperator
> > > > kafkaSinglePortStringInputOperator =
> > > > > new KafkaSinglePortStringInputOperator() ;
> > > > > DeserializeJSON deserializeJSON = new DeserializeJSON();
> > > > > RedisJoin redisJoin = new RedisJoin() ;
> > > > > CampaignProcessor campaignProcessor = new CampaignProcessor()
> ;
> > > > >
> > > > > kafkaInput.outputPort.connect(deserializeJSON.input);
> > > > > filterFields.output.connect(redisJoin.input) ;
> > > > > redisJoin.output.connect(campaignProcessor.input) ;
> > > > >
> > > > > dag.add(kafkaInput);
> > > > > -----------------------------------------
> > > > >
> > > > > Things to note : Name can be generated internally or user can
> > specify a
> > > > > name property of the operator ( like Dag.AddOperator("name", ...) )
> > > > >
> > > > > Let me know your thoughts.
> > > > >
> > > > > Thanks
> > > > >
> > > >
> > >
> >
>
Re: Proposal: Variation for the Apex Api
Posted by Sandesh Hegde <sa...@datatorrent.com>.
Hello Chinmay,
Idea is to follow natural order. Suppose you want to set attribute on a
port, you will have to write something like below
dag.setInputPortAttribute(deserializeJSON.input,
Context.PortContext.PARTITION_PARALLEL, true);
How about setting it directly on the ports?
deserializeJSON.input.setAttribute(Context.PortContext.PARTITION_PARALLEL,
true));
Same thing goes for attributes of the Operators.
These new set of APIs will support all the features of our
existing API, but in a more natural way.
It is not similar to High-Level API, it is just moonlighting idea about our
existing API.
Thanks
On Sun, Apr 10, 2016 at 11:20 PM Chinmay Kolhatkar <ch...@apache.org>
wrote:
> I feel that adding all operators to dag makes it explicit for a developer
> who is looking at the code for the first time.
>
> Besides, one might want to do dag.getMeta on operators and ports to set
> some attributes.
> I'm not sure how much explicit will that be if we bypass adding operators
> to the DAG.
>
> Though, I totally agree with the intention of simplifying the APIs for
> which the proposal from Siyuan in another mail is a nice first step.
>
> Regards,
> Chinmay.
>
>
>
> On Mon, Apr 11, 2016 at 10:48 AM, Sandesh Hegde <sa...@datatorrent.com>
> wrote:
>
> > Yes, that is the idea.
> >
> > On Sun, Apr 10, 2016, 9:28 PM Sandeep Deshmukh <sa...@datatorrent.com>
> > wrote:
> >
> > > Are you suggesting we add only the input adapters to the dag and then
> all
> > > the connected ones are added by simply traversing the DAG?
> > >
> > > Regards,
> > > Sandeep
> > >
> > > On Fri, Apr 8, 2016 at 11:06 PM, Sandesh Hegde <
> sandesh@datatorrent.com>
> > > wrote:
> > >
> > > > Hello Team,
> > > >
> > > > As we were talking about High Level api in another thread, here is a
> > > small
> > > > variation proposal for our existing apis,
> > > >
> > > > Take look at the current code and the proposed code
> > > >
> > > > Current Style
> > > > ---------------------------
> > > > KafkaSinglePortStringInputOperator kafkaInput =
> > > > dag.addOperator("kafkaInput", new
> > KafkaSinglePortStringInputOperator());
> > > > DeserializeJSON deserializeJSON = dag.addOperator("deserialize",
> > new
> > > > DeserializeJSON());
> > > > RedisJoin redisJoin = dag.addOperator("redisJoin", new
> > RedisJoin());
> > > > CampaignProcessorWithApexWindow campaignProcessor =
> > > > dag.addOperator("campaignProcessor", new
> > > > CampaignProcessorWithApexWindow());
> > > >
> > > > // Connect the Ports in the Operators
> > > > dag.addStream("deserialize", kafkaInput.outputPort,
> > > > deserializeJSON.input) ;
> > > > dag.addStream("redisJoin", filterFields.output, redisJoin.input)
> ;
> > > > dag.addStream("output", redisJoin.output,
> campaignProcessor.input);
> > > > -------------------------------------
> > > >
> > > >
> > > > Proposed Change ( Just create Operators and connect the output port
> to
> > > > input ports and then add the input operators to DAG )
> > > > -------------------------------------------
> > > > KafkaSinglePortStringInputOperator
> > > kafkaSinglePortStringInputOperator =
> > > > new KafkaSinglePortStringInputOperator() ;
> > > > DeserializeJSON deserializeJSON = new DeserializeJSON();
> > > > RedisJoin redisJoin = new RedisJoin() ;
> > > > CampaignProcessor campaignProcessor = new CampaignProcessor() ;
> > > >
> > > > kafkaInput.outputPort.connect(deserializeJSON.input);
> > > > filterFields.output.connect(redisJoin.input) ;
> > > > redisJoin.output.connect(campaignProcessor.input) ;
> > > >
> > > > dag.add(kafkaInput);
> > > > -----------------------------------------
> > > >
> > > > Things to note : Name can be generated internally or user can
> specify a
> > > > name property of the operator ( like Dag.AddOperator("name", ...) )
> > > >
> > > > Let me know your thoughts.
> > > >
> > > > Thanks
> > > >
> > >
> >
>
Re: Proposal: Variation for the Apex Api
Posted by Chinmay Kolhatkar <ch...@apache.org>.
I feel that adding all operators to dag makes it explicit for a developer
who is looking at the code for the first time.
Besides, one might want to do dag.getMeta on operators and ports to set
some attributes.
I'm not sure how much explicit will that be if we bypass adding operators
to the DAG.
Though, I totally agree with the intention of simplifying the APIs for
which the proposal from Siyuan in another mail is a nice first step.
Regards,
Chinmay.
On Mon, Apr 11, 2016 at 10:48 AM, Sandesh Hegde <sa...@datatorrent.com>
wrote:
> Yes, that is the idea.
>
> On Sun, Apr 10, 2016, 9:28 PM Sandeep Deshmukh <sa...@datatorrent.com>
> wrote:
>
> > Are you suggesting we add only the input adapters to the dag and then all
> > the connected ones are added by simply traversing the DAG?
> >
> > Regards,
> > Sandeep
> >
> > On Fri, Apr 8, 2016 at 11:06 PM, Sandesh Hegde <sa...@datatorrent.com>
> > wrote:
> >
> > > Hello Team,
> > >
> > > As we were talking about High Level api in another thread, here is a
> > small
> > > variation proposal for our existing apis,
> > >
> > > Take look at the current code and the proposed code
> > >
> > > Current Style
> > > ---------------------------
> > > KafkaSinglePortStringInputOperator kafkaInput =
> > > dag.addOperator("kafkaInput", new
> KafkaSinglePortStringInputOperator());
> > > DeserializeJSON deserializeJSON = dag.addOperator("deserialize",
> new
> > > DeserializeJSON());
> > > RedisJoin redisJoin = dag.addOperator("redisJoin", new
> RedisJoin());
> > > CampaignProcessorWithApexWindow campaignProcessor =
> > > dag.addOperator("campaignProcessor", new
> > > CampaignProcessorWithApexWindow());
> > >
> > > // Connect the Ports in the Operators
> > > dag.addStream("deserialize", kafkaInput.outputPort,
> > > deserializeJSON.input) ;
> > > dag.addStream("redisJoin", filterFields.output, redisJoin.input) ;
> > > dag.addStream("output", redisJoin.output, campaignProcessor.input);
> > > -------------------------------------
> > >
> > >
> > > Proposed Change ( Just create Operators and connect the output port to
> > > input ports and then add the input operators to DAG )
> > > -------------------------------------------
> > > KafkaSinglePortStringInputOperator
> > kafkaSinglePortStringInputOperator =
> > > new KafkaSinglePortStringInputOperator() ;
> > > DeserializeJSON deserializeJSON = new DeserializeJSON();
> > > RedisJoin redisJoin = new RedisJoin() ;
> > > CampaignProcessor campaignProcessor = new CampaignProcessor() ;
> > >
> > > kafkaInput.outputPort.connect(deserializeJSON.input);
> > > filterFields.output.connect(redisJoin.input) ;
> > > redisJoin.output.connect(campaignProcessor.input) ;
> > >
> > > dag.add(kafkaInput);
> > > -----------------------------------------
> > >
> > > Things to note : Name can be generated internally or user can specify a
> > > name property of the operator ( like Dag.AddOperator("name", ...) )
> > >
> > > Let me know your thoughts.
> > >
> > > Thanks
> > >
> >
>
Re: Proposal: Variation for the Apex Api
Posted by Sandesh Hegde <sa...@datatorrent.com>.
Yes, that is the idea.
On Sun, Apr 10, 2016, 9:28 PM Sandeep Deshmukh <sa...@datatorrent.com>
wrote:
> Are you suggesting we add only the input adapters to the dag and then all
> the connected ones are added by simply traversing the DAG?
>
> Regards,
> Sandeep
>
> On Fri, Apr 8, 2016 at 11:06 PM, Sandesh Hegde <sa...@datatorrent.com>
> wrote:
>
> > Hello Team,
> >
> > As we were talking about High Level api in another thread, here is a
> small
> > variation proposal for our existing apis,
> >
> > Take look at the current code and the proposed code
> >
> > Current Style
> > ---------------------------
> > KafkaSinglePortStringInputOperator kafkaInput =
> > dag.addOperator("kafkaInput", new KafkaSinglePortStringInputOperator());
> > DeserializeJSON deserializeJSON = dag.addOperator("deserialize", new
> > DeserializeJSON());
> > RedisJoin redisJoin = dag.addOperator("redisJoin", new RedisJoin());
> > CampaignProcessorWithApexWindow campaignProcessor =
> > dag.addOperator("campaignProcessor", new
> > CampaignProcessorWithApexWindow());
> >
> > // Connect the Ports in the Operators
> > dag.addStream("deserialize", kafkaInput.outputPort,
> > deserializeJSON.input) ;
> > dag.addStream("redisJoin", filterFields.output, redisJoin.input) ;
> > dag.addStream("output", redisJoin.output, campaignProcessor.input);
> > -------------------------------------
> >
> >
> > Proposed Change ( Just create Operators and connect the output port to
> > input ports and then add the input operators to DAG )
> > -------------------------------------------
> > KafkaSinglePortStringInputOperator
> kafkaSinglePortStringInputOperator =
> > new KafkaSinglePortStringInputOperator() ;
> > DeserializeJSON deserializeJSON = new DeserializeJSON();
> > RedisJoin redisJoin = new RedisJoin() ;
> > CampaignProcessor campaignProcessor = new CampaignProcessor() ;
> >
> > kafkaInput.outputPort.connect(deserializeJSON.input);
> > filterFields.output.connect(redisJoin.input) ;
> > redisJoin.output.connect(campaignProcessor.input) ;
> >
> > dag.add(kafkaInput);
> > -----------------------------------------
> >
> > Things to note : Name can be generated internally or user can specify a
> > name property of the operator ( like Dag.AddOperator("name", ...) )
> >
> > Let me know your thoughts.
> >
> > Thanks
> >
>
Re: Proposal: Variation for the Apex Api
Posted by Sandeep Deshmukh <sa...@datatorrent.com>.
Are you suggesting we add only the input adapters to the dag and then all
the connected ones are added by simply traversing the DAG?
Regards,
Sandeep
On Fri, Apr 8, 2016 at 11:06 PM, Sandesh Hegde <sa...@datatorrent.com>
wrote:
> Hello Team,
>
> As we were talking about High Level api in another thread, here is a small
> variation proposal for our existing apis,
>
> Take look at the current code and the proposed code
>
> Current Style
> ---------------------------
> KafkaSinglePortStringInputOperator kafkaInput =
> dag.addOperator("kafkaInput", new KafkaSinglePortStringInputOperator());
> DeserializeJSON deserializeJSON = dag.addOperator("deserialize", new
> DeserializeJSON());
> RedisJoin redisJoin = dag.addOperator("redisJoin", new RedisJoin());
> CampaignProcessorWithApexWindow campaignProcessor =
> dag.addOperator("campaignProcessor", new
> CampaignProcessorWithApexWindow());
>
> // Connect the Ports in the Operators
> dag.addStream("deserialize", kafkaInput.outputPort,
> deserializeJSON.input) ;
> dag.addStream("redisJoin", filterFields.output, redisJoin.input) ;
> dag.addStream("output", redisJoin.output, campaignProcessor.input);
> -------------------------------------
>
>
> Proposed Change ( Just create Operators and connect the output port to
> input ports and then add the input operators to DAG )
> -------------------------------------------
> KafkaSinglePortStringInputOperator kafkaSinglePortStringInputOperator =
> new KafkaSinglePortStringInputOperator() ;
> DeserializeJSON deserializeJSON = new DeserializeJSON();
> RedisJoin redisJoin = new RedisJoin() ;
> CampaignProcessor campaignProcessor = new CampaignProcessor() ;
>
> kafkaInput.outputPort.connect(deserializeJSON.input);
> filterFields.output.connect(redisJoin.input) ;
> redisJoin.output.connect(campaignProcessor.input) ;
>
> dag.add(kafkaInput);
> -----------------------------------------
>
> Things to note : Name can be generated internally or user can specify a
> name property of the operator ( like Dag.AddOperator("name", ...) )
>
> Let me know your thoughts.
>
> Thanks
>