You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@storm.apache.org by "Grant Overby (groverby)" <gr...@cisco.com> on 2015/09/24 18:35:21 UTC
Tident: fork DAG without partitioning?
I have a trident topology where a portion of the DAG looks like this:
partition - - - filterA - - - eachA
/
stream - - - -
\
partition - - - filterB - - - eachB
I believe with the above DAG, each tuple will be sent down both sides of the fork. Approximately half will be filtered out by each filter. Thus, forcing half my tuples to cross the partition only to get dropped afterwards.
Can the DAG be constructed like the following. If so, how do I define the topology?
filterA - - - partition - - - eachA
/
stream - - - -
\
filterB - - - partition - - - eachB
The following doesn't appear to work. filterA and filterB don't receive tuples.
Stream stream = ...;
stream.parallelismHint(N);
stream
.each(new Fields(...), filterA)
.localOrShuffle()
.each(new Fields(...), eachA, new Fields(...))
;
stream
.each(new Fields(...), filterB)
.localOrShuffle()
.each(new Fields(...), eachB, new Fields(...))
;
[http://www.cisco.com/web/europe/images/email/signature/est2014/logo_06.png?ct=1398192119726]
Grant Overby
Software Engineer
Cisco.com<http://www.cisco.com/>
groverby@cisco.com<ma...@cisco.com>
Mobile: 865 724 4910
[http://www.cisco.com/assets/swa/img/thinkbeforeyouprint.gif] Think before you print.
This email may contain confidential and privileged material for the sole use of the intended recipient. Any review, use, distribution or disclosure by others is strictly prohibited. If you are not the intended recipient (or authorized to receive for the recipient), please contact the sender by reply email and delete all copies of this message.
Please click here<http://www.cisco.com/web/about/doing_business/legal/cri/index.html> for Company Registration Information.
Re: Tident: fork DAG without partitioning?
Posted by "P. Taylor Goetz" <pt...@gmail.com>.
Hi Grant,
I think the fluent API is tripping you up a bit, but what you are trying to do is possible.
Stream rootStream = …;
Stream streamA = rootStream.each(new Fields(…), filterA);
streamA.localOrShuffle()
.each(new Fields(…), eachA, new Fields(…));
Stream streamB = rootStream.each(new Fields(…), new Negate(filterA));
streamB.localOrShuffle()
.each(new Fields(…), eachB, new Fields(…));
In streamB I used the Negate filter — all that does is invert an existing filter. That way you don’t have to write two filters that are just the opposite of one another.
-Taylor
> On Sep 24, 2015, at 12:35 PM, Grant Overby (groverby) <gr...@cisco.com> wrote:
>
> I have a trident topology where a portion of the DAG looks like this:
>
>
> partition - - - filterA - - - eachA
> /
> stream - - - -
> \
> partition - - - filterB - - - eachB
>
>
>
> I believe with the above DAG, each tuple will be sent down both sides of the fork. Approximately half will be filtered out by each filter. Thus, forcing half my tuples to cross the partition only to get dropped afterwards.
>
> Can the DAG be constructed like the following. If so, how do I define the topology?
>
> filterA - - - partition - - - eachA
> /
> stream - - - -
> \
> filterB - - - partition - - - eachB
>
>
>
>
> The following doesn’t appear to work. filterA and filterB don’t receive tuples.
>
> Stream stream = …;
>
> stream.parallelismHint(N);
>
> stream
> .each(new Fields(…), filterA)
> .localOrShuffle()
> .each(new Fields(…), eachA, new Fields(…))
> ;
>
> stream
> .each(new Fields(…), filterB)
> .localOrShuffle()
> .each(new Fields(…), eachB, new Fields(…))
> ;
>
>
>
>
>
>
>
> Grant Overby
> Software Engineer
> Cisco.com <http://www.cisco.com/>
> groverby@cisco.com <ma...@cisco.com>
> Mobile: 865 724 4910
>
>
> Think before you print.
> This email may contain confidential and privileged material for the sole use of the intended recipient. Any review, use, distribution or disclosure by others is strictly prohibited. If you are not the intended recipient (or authorized to receive for the recipient), please contact the sender by reply email and delete all copies of this message.
>
> Please click here <http://www.cisco.com/web/about/doing_business/legal/cri/index.html> for Company Registration Information.
>
>
>
>
Re: Tident: fork DAG without partitioning?
Posted by Tom Brown <to...@gmail.com>.
Could you have a combined filterAB that adds a new field, "which" populated
with "a" or "b", and then partition the stream based on that field?
--Tom
On Thursday, September 24, 2015, Grant Overby (groverby) <gr...@cisco.com>
wrote:
> Yes
>
> *Grant Overby*
> Software Engineer
> Cisco.com <http://www.cisco.com/>
> groverby@cisco.com <javascript:_e(%7B%7D,'cvml','groverby@cisco.com');>
> Mobile: *865 724 4910*
>
>
>
> Think before you print.
>
> This email may contain confidential and privileged material for the sole
> use of the intended recipient. Any review, use, distribution or disclosure
> by others is strictly prohibited. If you are not the intended recipient (or
> authorized to receive for the recipient), please contact the sender by
> reply email and delete all copies of this message.
>
> Please click here
> <http://www.cisco.com/web/about/doing_business/legal/cri/index.html> for
> Company Registration Information.
>
>
>
>
> From: Tom Brown <tombrown52@gmail.com
> <javascript:_e(%7B%7D,'cvml','tombrown52@gmail.com');>>
> Reply-To: "user@storm.apache.org
> <javascript:_e(%7B%7D,'cvml','user@storm.apache.org');>" <
> user@storm.apache.org
> <javascript:_e(%7B%7D,'cvml','user@storm.apache.org');>>
> Date: Thursday, September 24, 2015 at 1:04 PM
> To: "user@storm.apache.org
> <javascript:_e(%7B%7D,'cvml','user@storm.apache.org');>" <
> user@storm.apache.org
> <javascript:_e(%7B%7D,'cvml','user@storm.apache.org');>>
> Subject: Re: Tident: fork DAG without partitioning?
>
> Are your filters such that a tuple will pass either filterA, or filterB,
> but never both?
>
> --Tom
>
> On Thu, Sep 24, 2015 at 10:35 AM, Grant Overby (groverby) <
> groverby@cisco.com <javascript:_e(%7B%7D,'cvml','groverby@cisco.com');>>
> wrote:
>
>> I have a trident topology where a portion of the DAG looks like this:
>>
>>
>> partition - - - filterA - - - eachA
>> /
>> stream - - - -
>> \
>> partition - - - filterB - - - eachB
>>
>>
>>
>> I believe with the above DAG, each tuple will be sent down both sides of
>> the fork. Approximately half will be filtered out by each filter. Thus,
>> forcing half my tuples to cross the partition only to get dropped
>> afterwards.
>>
>> Can the DAG be constructed like the following. If so, how do I define the
>> topology?
>>
>> filterA - - - partition - - - eachA
>> /
>> stream - - - -
>> \
>> filterB - - - partition - - - eachB
>>
>>
>>
>>
>> The following doesn’t appear to work. filterA and filterB don’t receive
>> tuples.
>>
>> Stream stream = …;
>>
>> stream.parallelismHint(N);
>>
>> stream
>> .each(new Fields(…), filterA)
>> .localOrShuffle()
>> .each(new Fields(…), eachA, new Fields(…))
>> ;
>>
>> stream
>> .each(new Fields(…), filterB)
>> .localOrShuffle()
>> .each(new Fields(…), eachB, new Fields(…))
>> ;
>>
>>
>>
>>
>>
>>
>> *Grant Overby*
>> Software Engineer
>> Cisco.com <http://www.cisco.com/>
>> groverby@cisco.com <javascript:_e(%7B%7D,'cvml','groverby@cisco.com');>
>> Mobile: *865 724 4910 <865%20724%204910>*
>>
>>
>>
>> Think before you print.
>>
>> This email may contain confidential and privileged material for the sole
>> use of the intended recipient. Any review, use, distribution or disclosure
>> by others is strictly prohibited. If you are not the intended recipient (or
>> authorized to receive for the recipient), please contact the sender by
>> reply email and delete all copies of this message.
>>
>> Please click here
>> <http://www.cisco.com/web/about/doing_business/legal/cri/index.html> for
>> Company Registration Information.
>>
>>
>>
>>
>
Re: Tident: fork DAG without partitioning?
Posted by "Grant Overby (groverby)" <gr...@cisco.com>.
Yes
[http://www.cisco.com/web/europe/images/email/signature/est2014/logo_06.png?ct=1398192119726]
Grant Overby
Software Engineer
Cisco.com<http://www.cisco.com/>
groverby@cisco.com<ma...@cisco.com>
Mobile: 865 724 4910
[http://www.cisco.com/assets/swa/img/thinkbeforeyouprint.gif] Think before you print.
This email may contain confidential and privileged material for the sole use of the intended recipient. Any review, use, distribution or disclosure by others is strictly prohibited. If you are not the intended recipient (or authorized to receive for the recipient), please contact the sender by reply email and delete all copies of this message.
Please click here<http://www.cisco.com/web/about/doing_business/legal/cri/index.html> for Company Registration Information.
From: Tom Brown <to...@gmail.com>>
Reply-To: "user@storm.apache.org<ma...@storm.apache.org>" <us...@storm.apache.org>>
Date: Thursday, September 24, 2015 at 1:04 PM
To: "user@storm.apache.org<ma...@storm.apache.org>" <us...@storm.apache.org>>
Subject: Re: Tident: fork DAG without partitioning?
Are your filters such that a tuple will pass either filterA, or filterB, but never both?
--Tom
On Thu, Sep 24, 2015 at 10:35 AM, Grant Overby (groverby) <gr...@cisco.com>> wrote:
I have a trident topology where a portion of the DAG looks like this:
partition - - - filterA - - - eachA
/
stream - - - -
\
partition - - - filterB - - - eachB
I believe with the above DAG, each tuple will be sent down both sides of the fork. Approximately half will be filtered out by each filter. Thus, forcing half my tuples to cross the partition only to get dropped afterwards.
Can the DAG be constructed like the following. If so, how do I define the topology?
filterA - - - partition - - - eachA
/
stream - - - -
\
filterB - - - partition - - - eachB
The following doesn't appear to work. filterA and filterB don't receive tuples.
Stream stream = ...;
stream.parallelismHint(N);
stream
.each(new Fields(...), filterA)
.localOrShuffle()
.each(new Fields(...), eachA, new Fields(...))
;
stream
.each(new Fields(...), filterB)
.localOrShuffle()
.each(new Fields(...), eachB, new Fields(...))
;
[http://www.cisco.com/web/europe/images/email/signature/est2014/logo_06.png?ct=1398192119726]
Grant Overby
Software Engineer
Cisco.com<http://www.cisco.com/>
groverby@cisco.com<ma...@cisco.com>
Mobile: 865 724 4910<tel:865%20724%204910>
[http://www.cisco.com/assets/swa/img/thinkbeforeyouprint.gif] Think before you print.
This email may contain confidential and privileged material for the sole use of the intended recipient. Any review, use, distribution or disclosure by others is strictly prohibited. If you are not the intended recipient (or authorized to receive for the recipient), please contact the sender by reply email and delete all copies of this message.
Please click here<http://www.cisco.com/web/about/doing_business/legal/cri/index.html> for Company Registration Information.
Re: Tident: fork DAG without partitioning?
Posted by Tom Brown <to...@gmail.com>.
Are your filters such that a tuple will pass either filterA, or filterB,
but never both?
--Tom
On Thu, Sep 24, 2015 at 10:35 AM, Grant Overby (groverby) <
groverby@cisco.com> wrote:
> I have a trident topology where a portion of the DAG looks like this:
>
>
> partition - - - filterA - - - eachA
> /
> stream - - - -
> \
> partition - - - filterB - - - eachB
>
>
>
> I believe with the above DAG, each tuple will be sent down both sides of
> the fork. Approximately half will be filtered out by each filter. Thus,
> forcing half my tuples to cross the partition only to get dropped
> afterwards.
>
> Can the DAG be constructed like the following. If so, how do I define the
> topology?
>
> filterA - - - partition - - - eachA
> /
> stream - - - -
> \
> filterB - - - partition - - - eachB
>
>
>
>
> The following doesn’t appear to work. filterA and filterB don’t receive
> tuples.
>
> Stream stream = …;
>
> stream.parallelismHint(N);
>
> stream
> .each(new Fields(…), filterA)
> .localOrShuffle()
> .each(new Fields(…), eachA, new Fields(…))
> ;
>
> stream
> .each(new Fields(…), filterB)
> .localOrShuffle()
> .each(new Fields(…), eachB, new Fields(…))
> ;
>
>
>
>
>
>
> *Grant Overby*
> Software Engineer
> Cisco.com <http://www.cisco.com/>
> groverby@cisco.com
> Mobile: *865 724 4910 <865%20724%204910>*
>
>
>
> Think before you print.
>
> This email may contain confidential and privileged material for the sole
> use of the intended recipient. Any review, use, distribution or disclosure
> by others is strictly prohibited. If you are not the intended recipient (or
> authorized to receive for the recipient), please contact the sender by
> reply email and delete all copies of this message.
>
> Please click here
> <http://www.cisco.com/web/about/doing_business/legal/cri/index.html> for
> Company Registration Information.
>
>
>
>